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 8b3b0d330f [ISSUE #6461] Follow advertised WebSocket master addresses
(#7661)
8b3b0d330f is described below
commit 8b3b0d330fb566e2b99bc674cada10c7224a37df
Author: redyyang <[email protected]>
AuthorDate: Mon Oct 19 09:34:15 2026 +0800
[ISSUE #6461] Follow advertised WebSocket master addresses (#7661)
* fix(sync): follow advertised websocket master addresses
* test(e2e): use local HTTP fixture for wait helper checks
* test(e2e): use a local upstream for storage and cluster routing
* test(e2e): retain RocketMQ logs produced before queue assignment
* test(e2e): await discovered Spring Cloud route readiness
* test(e2e): match the Spring Cloud controller response
* test(e2e): explicitly assign RocketMQ log queues after topic creation
---
.github/workflows/e2e-k8s.yml | 2 +
.../shenyu-e2e-case/compose/fixtures/httpbin.py | 57 ++++
.../shenyu-e2e-case/compose/fixtures/httpbin.yml | 31 +++
.../compose/fixtures/test_httpbin.py | 97 +++++++
.../compose/storage/shenyu-storage-h2.yml | 6 +
.../compose/storage/shenyu-storage-mysql.yml | 6 +
.../compose/storage/shenyu-storage-opengauss.yml | 6 +
.../compose/storage/shenyu-storage-postgres.yml | 6 +
.../sync/shenyu-sync-websocket-cluster-jdbc.yml | 6 +
.../shenyu-sync-websocket-cluster-zookeeper.yml | 6 +
.../logging/rocketmq/DividePluginCases.java | 75 +++--
.../rocketmq/RocketMQLogConsumptionTest.java | 94 +++++++
.../shenyu-e2e-case-spring-cloud/pom.xml | 9 +
.../springcloud/SpringCloudPluginTest.java | 28 +-
.../springcloud/SpringCloudRouteReadinessTest.java | 57 ++++
.../e2e/engine/function/WaitForHelperTest.java | 69 ++++-
.../data/websocket/WebsocketSyncDataService.java | 36 ++-
.../websocket/client/ShenyuWebsocketClient.java | 12 +-
.../websocket/WebsocketMasterRedirectTest.java | 310 +++++++++++++++++++++
.../client/ShenyuWebsocketClientTest.java | 14 +
20 files changed, 882 insertions(+), 45 deletions(-)
diff --git a/.github/workflows/e2e-k8s.yml b/.github/workflows/e2e-k8s.yml
index ea72398f4c..7788f34f70 100644
--- a/.github/workflows/e2e-k8s.yml
+++ b/.github/workflows/e2e-k8s.yml
@@ -63,6 +63,8 @@ jobs:
submodules: true
- name: Test conditional disk cleanup
run: bash .github/scripts/free-disk-space-test.sh
+ - name: Test local HTTP upstream fixture and Compose wiring
+ run: python3 -B
shenyu-e2e/shenyu-e2e-case/compose/fixtures/test_httpbin.py
- uses: ./actions/paths-filter
id: filter
with:
diff --git a/shenyu-e2e/shenyu-e2e-case/compose/fixtures/httpbin.py
b/shenyu-e2e/shenyu-e2e-case/compose/fixtures/httpbin.py
new file mode 100644
index 0000000000..08c291145c
--- /dev/null
+++ b/shenyu-e2e/shenyu-e2e-case/compose/fixtures/httpbin.py
@@ -0,0 +1,57 @@
+# 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.
+
+"""Local upstream for the Divide URI and HTTP method routing scenarios."""
+
+import json
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from urllib.parse import parse_qs, urlsplit
+
+
+class HttpbinHandler(BaseHTTPRequestHandler):
+ protocol_version = "HTTP/1.1"
+
+ def respond(self):
+ path = urlsplit(self.path)
+ data = self.rfile.read(int(self.headers.get("Content-Length", "0")))
+ status = 200
+ if path.path == "/status/200":
+ body = b""
+ elif path.path == "/anything" or path.path.startswith("/anything/"):
+ body = json.dumps({
+ "method": self.command,
+ "url": self.path,
+ "args": parse_qs(path.query),
+ "data": data.decode("utf-8"),
+ }).encode("utf-8")
+ else:
+ status = 404
+ body = b'{"error":"unknown fixture endpoint"}'
+ self.send_response(status)
+ self.send_header("Content-Type", "application/json")
+ self.send_header("Content-Length", str(len(body)))
+ self.end_headers()
+ if self.command != "HEAD":
+ self.wfile.write(body)
+
+ do_GET = respond
+ do_POST = respond
+ do_PUT = respond
+ do_DELETE = respond
+ do_HEAD = respond
+
+
+if __name__ == "__main__":
+ ThreadingHTTPServer(("0.0.0.0", 80), HttpbinHandler).serve_forever()
diff --git a/shenyu-e2e/shenyu-e2e-case/compose/fixtures/httpbin.yml
b/shenyu-e2e/shenyu-e2e-case/compose/fixtures/httpbin.yml
new file mode 100644
index 0000000000..762d031d12
--- /dev/null
+++ b/shenyu-e2e/shenyu-e2e-case/compose/fixtures/httpbin.yml
@@ -0,0 +1,31 @@
+# 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.
+
+services:
+ httpbin:
+ image: python:3.12-alpine
+ command: ["python", "-B", "-u", "/opt/httpbin.py"]
+ volumes:
+ - ./httpbin.py:/opt/httpbin.py:ro
+ healthcheck:
+ test: ["CMD", "python", "-c", "import urllib.request;
urllib.request.urlopen('http://127.0.0.1/anything', timeout=2)"]
+ interval: 2s
+ timeout: 3s
+ retries: 15
+ networks:
+ shenyu:
+ # Existing routing scenarios resolve their upstream on this test
network.
+ aliases:
+ - httpbin.org
diff --git a/shenyu-e2e/shenyu-e2e-case/compose/fixtures/test_httpbin.py
b/shenyu-e2e/shenyu-e2e-case/compose/fixtures/test_httpbin.py
new file mode 100644
index 0000000000..ec0741a1a6
--- /dev/null
+++ b/shenyu-e2e/shenyu-e2e-case/compose/fixtures/test_httpbin.py
@@ -0,0 +1,97 @@
+# 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.
+
+"""Exercise the upstream over HTTP and validate every consuming Compose
file."""
+
+import http.client
+import json
+from pathlib import Path
+import subprocess
+import threading
+import unittest
+
+from httpbin import HttpbinHandler, ThreadingHTTPServer
+
+
+class HttpbinTest(unittest.TestCase):
+ @classmethod
+ def setUpClass(cls):
+ cls.server = ThreadingHTTPServer(("127.0.0.1", 0), HttpbinHandler)
+ cls.thread = threading.Thread(target=cls.server.serve_forever)
+ cls.thread.start()
+
+ @classmethod
+ def tearDownClass(cls):
+ cls.server.shutdown()
+ cls.server.server_close()
+ cls.thread.join()
+
+ def setUp(self):
+ self.connection =
http.client.HTTPConnection(*self.server.server_address, timeout=3)
+
+ def tearDown(self):
+ self.connection.close()
+
+ def test_echo_methods_paths_and_request_bodies(self):
+ for method in ("GET", "POST", "PUT", "DELETE"):
+ for path in ("/anything", "/anything/xx", "/anything/xx/yy",
"/anything/xxx"):
+ with self.subTest(method=method, path=path):
+ self.connection.request(method, path + "?id=23",
body="request body")
+ response = self.connection.getresponse()
+ self.assertEqual(200, response.status)
+ body = json.loads(response.read())
+ self.assertEqual(method, body["method"])
+ self.assertEqual(path + "?id=23", body["url"])
+ self.assertEqual({"id": ["23"]}, body["args"])
+ self.assertEqual("request body", body["data"])
+
+ def test_status_200_has_empty_body_for_every_method(self):
+ for method in ("GET", "POST", "PUT", "DELETE", "HEAD"):
+ with self.subTest(method=method):
+ self.connection.request(method, "/status/200")
+ response = self.connection.getresponse()
+ self.assertEqual(200, response.status)
+ self.assertEqual(b"", response.read())
+
+ def test_unknown_path_is_not_a_success(self):
+ self.connection.request("GET", "/not-a-fixture")
+ response = self.connection.getresponse()
+ self.assertEqual(404, response.status)
+ response.read()
+
+ def test_compose_uses_healthy_local_upstream(self):
+ directory = Path(__file__).resolve().parent
+ compose = directory.parent
+ files = sorted((compose / "storage").glob("shenyu-storage-*.yml"))
+ files += sorted((compose /
"sync").glob("shenyu-sync-websocket-cluster-*.yml"))
+ self.assertEqual(6, len(files))
+ for path in files:
+ with self.subTest(compose=path.name):
+ result = subprocess.run(
+ ["docker", "compose", "-f", str(path), "config",
"--format", "json"],
+ check=True, capture_output=True, text=True,
+ )
+ services = json.loads(result.stdout)["services"]
+ fixture = services["httpbin"]
+ self.assertIn("httpbin.org",
fixture["networks"]["shenyu"]["aliases"])
+ self.assertEqual("service_healthy",
services["shenyu-bootstrap"]["depends_on"]["httpbin"]["condition"])
+ mount = next(v for v in fixture["volumes"] if v["target"] ==
"/opt/httpbin.py")
+ self.assertEqual(directory / "httpbin.py",
Path(mount["source"]))
+ self.assertTrue(mount["read_only"])
+ self.assertIn("/anything", fixture["healthcheck"]["test"][-1])
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/shenyu-e2e/shenyu-e2e-case/compose/storage/shenyu-storage-h2.yml
b/shenyu-e2e/shenyu-e2e-case/compose/storage/shenyu-storage-h2.yml
index c35ccd1dbe..a722a0501f 100644
--- a/shenyu-e2e/shenyu-e2e-case/compose/storage/shenyu-storage-h2.yml
+++ b/shenyu-e2e/shenyu-e2e-case/compose/storage/shenyu-storage-h2.yml
@@ -17,6 +17,10 @@
version: '3.9'
services:
+ httpbin:
+ extends:
+ file: ../fixtures/httpbin.yml
+ service: httpbin
shenyu-admin:
image: apache/shenyu-admin:latest
@@ -63,6 +67,8 @@ services:
start_period: 30s
restart: always
depends_on:
+ httpbin:
+ condition: service_healthy
shenyu-admin:
condition: service_healthy
networks:
diff --git
a/shenyu-e2e/shenyu-e2e-case/compose/storage/shenyu-storage-mysql.yml
b/shenyu-e2e/shenyu-e2e-case/compose/storage/shenyu-storage-mysql.yml
index 06f0b40476..2141c9d9a8 100644
--- a/shenyu-e2e/shenyu-e2e-case/compose/storage/shenyu-storage-mysql.yml
+++ b/shenyu-e2e/shenyu-e2e-case/compose/storage/shenyu-storage-mysql.yml
@@ -17,6 +17,10 @@
version: '3.9'
services:
+ httpbin:
+ extends:
+ file: ../fixtures/httpbin.yml
+ service: httpbin
shenyu-mysql:
image: mysql:8.0
container_name: shenyu-mysql
@@ -98,6 +102,8 @@ services:
start_period: 30s
restart: always
depends_on:
+ httpbin:
+ condition: service_healthy
shenyu-admin:
condition: service_healthy
networks:
diff --git
a/shenyu-e2e/shenyu-e2e-case/compose/storage/shenyu-storage-opengauss.yml
b/shenyu-e2e/shenyu-e2e-case/compose/storage/shenyu-storage-opengauss.yml
index d0c3b66228..53f62e6303 100644
--- a/shenyu-e2e/shenyu-e2e-case/compose/storage/shenyu-storage-opengauss.yml
+++ b/shenyu-e2e/shenyu-e2e-case/compose/storage/shenyu-storage-opengauss.yml
@@ -17,6 +17,10 @@
version: '3.9'
services:
+ httpbin:
+ extends:
+ file: ../fixtures/httpbin.yml
+ service: httpbin
shenyu-opengauss:
image: enmotech/opengauss-lite:5.0.1
container_name: shenyu-opengauss
@@ -97,6 +101,8 @@ services:
start_period: 30s
restart: always
depends_on:
+ httpbin:
+ condition: service_healthy
shenyu-admin:
condition: service_healthy
networks:
diff --git
a/shenyu-e2e/shenyu-e2e-case/compose/storage/shenyu-storage-postgres.yml
b/shenyu-e2e/shenyu-e2e-case/compose/storage/shenyu-storage-postgres.yml
index 29e3e87aca..147d78b699 100644
--- a/shenyu-e2e/shenyu-e2e-case/compose/storage/shenyu-storage-postgres.yml
+++ b/shenyu-e2e/shenyu-e2e-case/compose/storage/shenyu-storage-postgres.yml
@@ -17,6 +17,10 @@
version: '3.9'
services:
+ httpbin:
+ extends:
+ file: ../fixtures/httpbin.yml
+ service: httpbin
shenyu-postgres:
image: postgres:14
container_name: shenyu-postgres
@@ -97,6 +101,8 @@ services:
start_period: 30s
restart: always
depends_on:
+ httpbin:
+ condition: service_healthy
shenyu-admin:
condition: service_healthy
networks:
diff --git
a/shenyu-e2e/shenyu-e2e-case/compose/sync/shenyu-sync-websocket-cluster-jdbc.yml
b/shenyu-e2e/shenyu-e2e-case/compose/sync/shenyu-sync-websocket-cluster-jdbc.yml
index 85ac6bf35c..7142d5d8d5 100644
---
a/shenyu-e2e/shenyu-e2e-case/compose/sync/shenyu-sync-websocket-cluster-jdbc.yml
+++
b/shenyu-e2e/shenyu-e2e-case/compose/sync/shenyu-sync-websocket-cluster-jdbc.yml
@@ -17,6 +17,10 @@
version: '3.9'
services:
+ httpbin:
+ extends:
+ file: ../fixtures/httpbin.yml
+ service: httpbin
shenyu-mysql:
image: mysql:8.0
container_name: shenyu-mysql
@@ -133,6 +137,8 @@ services:
start_period: 30s
restart: always
depends_on:
+ httpbin:
+ condition: service_healthy
shenyu-admin-1:
condition: service_healthy
shenyu-admin-2:
diff --git
a/shenyu-e2e/shenyu-e2e-case/compose/sync/shenyu-sync-websocket-cluster-zookeeper.yml
b/shenyu-e2e/shenyu-e2e-case/compose/sync/shenyu-sync-websocket-cluster-zookeeper.yml
index 58d3d41b5b..79dea728be 100644
---
a/shenyu-e2e/shenyu-e2e-case/compose/sync/shenyu-sync-websocket-cluster-zookeeper.yml
+++
b/shenyu-e2e/shenyu-e2e-case/compose/sync/shenyu-sync-websocket-cluster-zookeeper.yml
@@ -17,6 +17,10 @@
version: '3.9'
services:
+ httpbin:
+ extends:
+ file: ../fixtures/httpbin.yml
+ service: httpbin
shenyu-zookeeper-cluster:
image: zookeeper:3.9
container_name: shenyu-zookeeper-cluster
@@ -158,6 +162,8 @@ services:
start_period: 30s
restart: always
depends_on:
+ httpbin:
+ condition: service_healthy
shenyu-admin-1:
condition: service_healthy
shenyu-admin-2:
diff --git
a/shenyu-e2e/shenyu-e2e-case/shenyu-e2e-case-logging-rocketmq/src/test/java/org/apache/shenyu/e2e/testcase/logging/rocketmq/DividePluginCases.java
b/shenyu-e2e/shenyu-e2e-case/shenyu-e2e-case-logging-rocketmq/src/test/java/org/apache/shenyu/e2e/testcase/logging/rocketmq/DividePluginCases.java
index 87ce0e4a96..a0cb0fdd6a 100644
---
a/shenyu-e2e/shenyu-e2e-case/shenyu-e2e-case-logging-rocketmq/src/test/java/org/apache/shenyu/e2e/testcase/logging/rocketmq/DividePluginCases.java
+++
b/shenyu-e2e/shenyu-e2e-case/shenyu-e2e-case-logging-rocketmq/src/test/java/org/apache/shenyu/e2e/testcase/logging/rocketmq/DividePluginCases.java
@@ -19,10 +19,11 @@ package org.apache.shenyu.e2e.testcase.logging.rocketmq;
import com.google.common.collect.Lists;
import io.restassured.http.Method;
-import org.apache.commons.collections.CollectionUtils;
-import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
-import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
-import
org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
+import org.apache.rocketmq.client.consumer.DefaultLitePullConsumer;
+import org.apache.rocketmq.client.exception.MQClientException;
+import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
+import org.apache.rocketmq.common.message.MessageQueue;
+import org.apache.rocketmq.common.protocol.ResponseCode;
import org.apache.shenyu.e2e.engine.scenario.ShenYuScenarioProvider;
import org.apache.shenyu.e2e.engine.scenario.specification.ScenarioSpec;
import
org.apache.shenyu.e2e.engine.scenario.specification.ShenYuBeforeEachSpec;
@@ -35,8 +36,12 @@ import org.junit.jupiter.api.Assertions;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.nio.charset.StandardCharsets;
import java.time.Duration;
+import java.util.Collection;
import java.util.List;
+import java.util.Objects;
+import java.util.UUID;
import java.util.concurrent.atomic.AtomicBoolean;
import static
org.apache.shenyu.e2e.engine.scenario.function.HttpCheckers.exists;
@@ -55,7 +60,7 @@ public class DividePluginCases implements
ShenYuScenarioProvider {
private static final String TEST = "/http/order/findById?id=123";
- private static final Duration LOG_CONSUME_TIMEOUT = Duration.ofSeconds(30);
+ private static final Duration LOG_CONSUME_TIMEOUT = Duration.ofSeconds(60);
private static final Logger LOG =
LoggerFactory.getLogger(DividePluginCases.class);
@@ -79,6 +84,42 @@ public class DividePluginCases implements
ShenYuScenarioProvider {
.build();
}
+ static void awaitRequestLog(final DefaultLitePullConsumer consumer, final
String endpoint) {
+ AtomicBoolean assigned = new AtomicBoolean();
+ await().alias("RocketMQ access log for " + endpoint)
+
.atMost(LOG_CONSUME_TIMEOUT).pollInterval(Duration.ofMillis(200))
+ .until(() -> {
+ if (!assigned.get()) {
+ Collection<MessageQueue> queues;
+ try {
+ queues = consumer.fetchMessageQueues(TOPIC);
+ } catch (MQClientException ex) {
+ Throwable cause = ex;
+ while (Objects.nonNull(cause)) {
+ if (cause instanceof MQClientException &&
((MQClientException) cause).getResponseCode() == ResponseCode.TOPIC_NOT_EXIST) {
+ return false;
+ }
+ cause = cause.getCause();
+ }
+ throw ex;
+ }
+ if (queues.isEmpty()) {
+ return false;
+ }
+ consumer.assign(queues);
+ for (MessageQueue queue : queues) {
+ consumer.seekToBegin(queue);
+ }
+ assigned.set(true);
+ }
+ return consumer.poll(200).stream().anyMatch(message -> {
+ String body = new String(message.getBody(),
StandardCharsets.UTF_8);
+ LOG.info("RocketMQ access log: {}", body);
+ return body.contains(endpoint);
+ });
+ });
+ }
+
private ShenYuScenarioSpec testRocketMQHello() {
return ShenYuScenarioSpec.builder()
.name("testRocketMQHello")
@@ -102,27 +143,17 @@ public class DividePluginCases implements
ShenYuScenarioProvider {
.caseSpec(
ShenYuCaseSpec.builder()
.add(request -> {
- AtomicBoolean isLog = new
AtomicBoolean(false);
- DefaultMQPushConsumer consumer = new
DefaultMQPushConsumer(CONSUMERGROUP);
+ String requestId =
UUID.randomUUID().toString();
+ String endpoint =
"/http/order/findById?id=23&e2e=" + requestId;
+ DefaultLitePullConsumer consumer = new
DefaultLitePullConsumer(CONSUMERGROUP + "-" + requestId);
try {
consumer.setNamesrvAddr(NAMESERVER);
- consumer.subscribe(TOPIC, "*");
-
consumer.registerMessageListener((MessageListenerConcurrently) (msgs,
consumeConcurrentlyContext) -> {
- LOG.info("Msg:{}", msgs);
- if
(CollectionUtils.isNotEmpty(msgs)) {
- msgs.forEach(e -> {
- if (new
String(e.getBody()).contains("/http/order/findById?id=23")) {
- isLog.set(true);
- }
- });
- }
- return
ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
- });
+
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
+ consumer.setAutoCommit(false);
consumer.start();
LOG.info("RocketMQ consumer started");
- request.request(Method.GET,
"/http/order/findById?id=23");
-
await().atMost(LOG_CONSUME_TIMEOUT).untilTrue(isLog);
- LOG.info("isLog.get():{}",
isLog.get());
+ request.request(Method.GET,
endpoint).then().statusCode(200);
+ awaitRequestLog(consumer, endpoint);
} catch (Exception e) {
LOG.error("error", e);
Assertions.fail("Failed to consume
RocketMQ access log", e);
diff --git
a/shenyu-e2e/shenyu-e2e-case/shenyu-e2e-case-logging-rocketmq/src/test/java/org/apache/shenyu/e2e/testcase/logging/rocketmq/RocketMQLogConsumptionTest.java
b/shenyu-e2e/shenyu-e2e-case/shenyu-e2e-case-logging-rocketmq/src/test/java/org/apache/shenyu/e2e/testcase/logging/rocketmq/RocketMQLogConsumptionTest.java
new file mode 100644
index 0000000000..d620ca3c1f
--- /dev/null
+++
b/shenyu-e2e/shenyu-e2e-case/shenyu-e2e-case-logging-rocketmq/src/test/java/org/apache/shenyu/e2e/testcase/logging/rocketmq/RocketMQLogConsumptionTest.java
@@ -0,0 +1,94 @@
+/*
+ * 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.e2e.testcase.logging.rocketmq;
+
+import org.apache.rocketmq.client.consumer.DefaultLitePullConsumer;
+import org.apache.rocketmq.client.exception.MQClientException;
+import org.apache.rocketmq.common.message.MessageExt;
+import org.apache.rocketmq.common.message.MessageQueue;
+import org.apache.rocketmq.common.protocol.ResponseCode;
+import org.junit.jupiter.api.Test;
+
+import java.nio.charset.StandardCharsets;
+import java.util.Collection;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class RocketMQLogConsumptionTest {
+
+ @Test
+ void rejectsNonTransientBrokerErrors() {
+ DefaultLitePullConsumer consumer = new
DefaultLitePullConsumer("denied-log-fixture") {
+ @Override
+ public Collection<MessageQueue> fetchMessageQueues(final String
topic) throws MQClientException {
+ throw new MQClientException(ResponseCode.NO_PERMISSION,
"access denied");
+ }
+ };
+
+ MQClientException exception = assertThrows(MQClientException.class, ()
-> DividePluginCases.awaitRequestLog(consumer, "/current-request"));
+ assertEquals(ResponseCode.NO_PERMISSION, exception.getResponseCode());
+ }
+
+ @Test
+ void waitsForTopicCreationAndRejectsUnrelatedLogs() {
+ AtomicInteger discoveries = new AtomicInteger();
+ AtomicInteger polls = new AtomicInteger();
+ AtomicInteger seeks = new AtomicInteger();
+ MessageQueue queue = new MessageQueue("shenyu-access-logging",
"broker", 0);
+ String endpoint = "/http/order/findById?id=23&e2e=current-request";
+ DefaultLitePullConsumer consumer = new
DefaultLitePullConsumer("log-fixture") {
+ @Override
+ public Collection<MessageQueue> fetchMessageQueues(final String
topic) throws MQClientException {
+ if (discoveries.incrementAndGet() == 1) {
+ throw new MQClientException("topic route pending", new
MQClientException(ResponseCode.TOPIC_NOT_EXIST, "topic is not published yet"));
+ }
+ return List.of(queue);
+ }
+
+ @Override
+ public void assign(final Collection<MessageQueue> queues) {
+ assertEquals(List.of(queue), queues);
+ }
+
+ @Override
+ public void seekToBegin(final MessageQueue assignedQueue) {
+ assertEquals(queue, assignedQueue);
+ seeks.incrementAndGet();
+ }
+
+ @Override
+ public List<MessageExt> poll(final long timeout) {
+ assertTrue(seeks.get() > 0);
+ MessageExt message = new MessageExt();
+ String uri = polls.incrementAndGet() == 1 ?
"/http/order/findById?id=23&e2e=old-request" : endpoint;
+ message.setBody(("{\"requestUri\":\"http://gateway"; + uri +
"\"}").getBytes(StandardCharsets.UTF_8));
+ return List.of(message);
+ }
+ };
+
+ DividePluginCases.awaitRequestLog(consumer, endpoint);
+
+ assertEquals(2, discoveries.get());
+ assertEquals(1, seeks.get());
+ assertEquals(2, polls.get());
+ }
+}
diff --git a/shenyu-e2e/shenyu-e2e-case/shenyu-e2e-case-spring-cloud/pom.xml
b/shenyu-e2e/shenyu-e2e-case/shenyu-e2e-case-spring-cloud/pom.xml
index 4ff46f9ed9..ef1d944bd2 100644
--- a/shenyu-e2e/shenyu-e2e-case/shenyu-e2e-case-spring-cloud/pom.xml
+++ b/shenyu-e2e/shenyu-e2e-case/shenyu-e2e-case-spring-cloud/pom.xml
@@ -28,4 +28,13 @@
<artifactId>shenyu-e2e-case-spring-cloud</artifactId>
+ <dependencies>
+ <dependency>
+ <groupId>org.awaitility</groupId>
+ <artifactId>awaitility</artifactId>
+ <version>${awaitility.version}</version>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+
</project>
diff --git
a/shenyu-e2e/shenyu-e2e-case/shenyu-e2e-case-spring-cloud/src/test/java/org/apache/shenyu/e2e/testcase/springcloud/SpringCloudPluginTest.java
b/shenyu-e2e/shenyu-e2e-case/shenyu-e2e-case-spring-cloud/src/test/java/org/apache/shenyu/e2e/testcase/springcloud/SpringCloudPluginTest.java
index d2847f639e..5166c418ed 100644
---
a/shenyu-e2e/shenyu-e2e-case/shenyu-e2e-case-spring-cloud/src/test/java/org/apache/shenyu/e2e/testcase/springcloud/SpringCloudPluginTest.java
+++
b/shenyu-e2e/shenyu-e2e-case/shenyu-e2e-case-spring-cloud/src/test/java/org/apache/shenyu/e2e/testcase/springcloud/SpringCloudPluginTest.java
@@ -20,7 +20,6 @@ package org.apache.shenyu.e2e.testcase.springcloud;
import org.apache.shenyu.e2e.client.WaitDataSync;
import org.apache.shenyu.e2e.client.admin.AdminClient;
import org.apache.shenyu.e2e.client.gateway.GatewayClient;
-import org.apache.shenyu.e2e.constant.Constants;
import org.apache.shenyu.e2e.engine.annotation.ShenYuScenario;
import org.apache.shenyu.e2e.engine.annotation.ShenYuTest;
import org.apache.shenyu.e2e.engine.scenario.specification.CaseSpec;
@@ -31,9 +30,11 @@ import org.junit.jupiter.api.BeforeEach;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.util.HashMap;
+import java.time.Duration;
import java.util.List;
-import java.util.Map;
+
+import static org.awaitility.Awaitility.await;
+import static org.hamcrest.Matchers.equalTo;
/**
* Testing spring-cloud plugin.
@@ -74,17 +75,20 @@ public class SpringCloudPluginTest {
List<SelectorCacheData> selectorCache =
gatewayClient.getSelectorCache();
LOG.info("selectorCache: {}", selectorCache);
- Map<String, String> reqBody = new HashMap<>();
- reqBody.put("pluginId", "5");
- reqBody.put("name", "divide");
- reqBody.put("enabled", "true");
- reqBody.put("role", "Proxy");
- reqBody.put("sort", "200");
- reqBody.put("config", "{\"enabled\":true}");
- reqBody.put("namespaceId",
Constants.SYS_DEFAULT_NAMESPACE_NAMESPACE_ID);
- adminClient.changePluginStatus("1801816010882822185", reqBody);
WaitDataSync.waitGatewayPluginUse(gatewayClient,
"org.apache.shenyu.plugin.divide.DividePlugin");
+ waitForRegisteredRoute(gatewayClient);
+ }
+ static void waitForRegisteredRoute(final GatewayClient gatewayClient) {
+ // Selector/rule synchronization can complete before Eureka publishes
the upstream.
+ await().alias("Spring Cloud discovery upstream routes through the
gateway")
+ .atMost(Duration.ofSeconds(90))
+ .pollInterval(Duration.ofMillis(200))
+ .untilAsserted(() ->
gatewayClient.getHttpRequesterSupplier().get()
+ .get("/springcloud/order/path/123/hahah")
+ .then().statusCode(200)
+ .body("id", equalTo("123"))
+ .body("name", equalTo("hello world spring cloud
restful: hahah")));
}
@ShenYuScenario(provider = SpringCloudPluginCases.class)
diff --git
a/shenyu-e2e/shenyu-e2e-case/shenyu-e2e-case-spring-cloud/src/test/java/org/apache/shenyu/e2e/testcase/springcloud/SpringCloudRouteReadinessTest.java
b/shenyu-e2e/shenyu-e2e-case/shenyu-e2e-case-spring-cloud/src/test/java/org/apache/shenyu/e2e/testcase/springcloud/SpringCloudRouteReadinessTest.java
new file mode 100644
index 0000000000..ee1adb4612
--- /dev/null
+++
b/shenyu-e2e/shenyu-e2e-case/shenyu-e2e-case-spring-cloud/src/test/java/org/apache/shenyu/e2e/testcase/springcloud/SpringCloudRouteReadinessTest.java
@@ -0,0 +1,57 @@
+/*
+ * 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.e2e.testcase.springcloud;
+
+import com.sun.net.httpserver.HttpServer;
+import org.apache.shenyu.e2e.client.gateway.GatewayClient;
+import org.junit.jupiter.api.Test;
+
+import java.net.InetSocketAddress;
+import java.nio.charset.StandardCharsets;
+import java.util.Properties;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class SpringCloudRouteReadinessTest {
+
+ @Test
+ void waitsForDiscoveredUpstreamAndExpectedResponse() throws Exception {
+ AtomicInteger requests = new AtomicInteger();
+ HttpServer server = HttpServer.create(new
InetSocketAddress("127.0.0.1", 0), 0);
+ server.createContext("/springcloud/order/path/123/hahah", exchange -> {
+ int attempt = requests.incrementAndGet();
+ byte[] body = (attempt < 3 ?
"{\"code\":-107,\"message\":\"upstream not ready\"}"
+ : "{\"id\":\"123\",\"name\":\"hello world spring cloud
restful: hahah\"}").getBytes(StandardCharsets.UTF_8);
+ exchange.getResponseHeaders().set("Content-Type",
"application/json");
+ exchange.sendResponseHeaders(attempt == 1 ? 503 : 200,
body.length);
+ try (var output = exchange.getResponseBody()) {
+ output.write(body);
+ }
+ });
+ server.start();
+ try {
+ GatewayClient gateway = new GatewayClient("readiness-test",
"gateway",
+ "http://127.0.0.1:"; + server.getAddress().getPort(), new
Properties());
+ SpringCloudPluginTest.waitForRegisteredRoute(gateway);
+ assertTrue(requests.get() >= 3);
+ } finally {
+ server.stop(0);
+ }
+ }
+}
diff --git
a/shenyu-e2e/shenyu-e2e-engine/src/test/java/org/apache/shenyu/e2e/engine/function/WaitForHelperTest.java
b/shenyu-e2e/shenyu-e2e-engine/src/test/java/org/apache/shenyu/e2e/engine/function/WaitForHelperTest.java
index 10220adb54..9f07f4a5bd 100644
---
a/shenyu-e2e/shenyu-e2e-engine/src/test/java/org/apache/shenyu/e2e/engine/function/WaitForHelperTest.java
+++
b/shenyu-e2e/shenyu-e2e-engine/src/test/java/org/apache/shenyu/e2e/engine/function/WaitForHelperTest.java
@@ -17,13 +17,23 @@
package org.apache.shenyu.e2e.engine.function;
-import org.apache.shenyu.e2e.engine.scenario.function.WaitForHelper;
+import com.sun.net.httpserver.HttpExchange;
+import com.sun.net.httpserver.HttpServer;
import io.restassured.builder.ResponseSpecBuilder;
import io.restassured.http.Method;
+import org.apache.shenyu.e2e.engine.scenario.function.WaitForHelper;
+import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.net.InetSocketAddress;
+import java.nio.charset.StandardCharsets;
import java.time.Duration;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import static io.restassured.RestAssured.given;
@@ -31,26 +41,73 @@ import static org.hamcrest.CoreMatchers.containsString;
import static org.hamcrest.CoreMatchers.equalTo;
public class WaitForHelperTest {
-
+
+ private final CountDownLatch delayedRequest = new CountDownLatch(1);
+
+ private final CountDownLatch releaseResponse = new CountDownLatch(1);
+
+ private HttpServer server;
+
+ @BeforeEach
+ void setUp() throws IOException {
+ server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
+ server.createContext("/get", exchange -> writeJson(exchange,
"{\"url\":\"" + exchange.getRequestURI() + "\"}"));
+ server.createContext("/delay/2", exchange -> {
+ delayedRequest.countDown();
+ try {
+ releaseResponse.await();
+ writeJson(exchange, "{\"delay\":2}");
+ } catch (InterruptedException ex) {
+ Thread.currentThread().interrupt();
+ exchange.close();
+ }
+ });
+ server.start();
+ // Initialize the HTTP client before measuring the helper's short
timeout.
+ given().baseUri(baseUri()).get("/get").then().statusCode(200);
+ }
+
+ @AfterEach
+ void tearDown() {
+ releaseResponse.countDown();
+ server.stop(0);
+ }
+
@Test
void testSuccess() throws TimeoutException {
new WaitForHelper().waitFor(
- () -> given().baseUri("https://postman-echo.com";).when(),
+ () -> given().baseUri(baseUri()).when(),
Method.GET,
"/get?foo1=bar1&foo2=bar2",
new ResponseSpecBuilder().expectBody("url",
containsString("/get")).build()
);
}
-
+
@Test
- void testTimeout() {
+ void testTimeout() throws InterruptedException {
Assertions.assertThrows(TimeoutException.class, () ->
new WaitForHelper(10, Duration.ofSeconds(1),
Duration.ofMillis(500)).waitFor(
- () -> given().baseUri("https://postman-echo.com";).when(),
+ () -> given().baseUri(baseUri()).when(),
Method.GET,
"/delay/2",
new ResponseSpecBuilder().expectBody("delay",
equalTo(2)).build()
)
);
+ Assertions.assertTrue(delayedRequest.await(2, TimeUnit.SECONDS));
+ }
+
+ private String baseUri() {
+ return "http://127.0.0.1:"; + server.getAddress().getPort();
+ }
+
+ private static void writeJson(final HttpExchange exchange, final String
body) throws IOException {
+ byte[] bytes = body.getBytes(StandardCharsets.UTF_8);
+ exchange.getResponseHeaders().set("Content-Type", "application/json");
+ exchange.sendResponseHeaders(200, bytes.length);
+ try (OutputStream output = exchange.getResponseBody()) {
+ output.write(bytes);
+ } finally {
+ exchange.close();
+ }
}
}
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 4bab04bf75..c2f32418a2 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
@@ -43,9 +43,11 @@ import
org.springframework.boot.autoconfigure.web.ServerProperties;
import java.net.URI;
import java.util.HashMap;
import java.util.Iterator;
+import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
+import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -150,6 +152,7 @@ public class WebsocketSyncDataService implements
SyncDataService {
clients.add(createClient(url));
}
}
+ Set<URI> masterUrls = new LinkedHashSet<>();
Iterator<ShenyuWebsocketClient> iterator = clients.iterator();
while (iterator.hasNext()) {
ShenyuWebsocketClient websocketClient = iterator.next();
@@ -166,11 +169,42 @@ public class WebsocketSyncDataService implements
SyncDataService {
return;
}
- if (!websocketClient.isConnectedToMaster()) {
+ // An open socket may still be waiting for its running-mode reply.
+ if (Objects.equals(runningMode, RunningModeEnum.CLUSTER.name()) &&
!websocketClient.isConnectedToMaster()) {
+ URI masterUri = parseMasterUri(websocketClient.getMasterUrl());
+ if (Objects.nonNull(masterUri) &&
!masterUri.equals(websocketClient.getURI())) {
+ masterUrls.add(masterUri);
+ }
websocketClient.nowClose();
iterator.remove();
}
}
+ for (URI masterUri : masterUrls) {
+ if (clients.stream().noneMatch(client ->
masterUri.equals(client.getURI()))) {
+ try {
+ clients.add(createClient(masterUri.toString()));
+ } catch (RuntimeException ex) {
+ LOG.warn("websocket connection to advertised master [{}]
failed", masterUri, ex);
+ }
+ }
+ }
+ }
+
+ private URI parseMasterUri(final String masterUrl) {
+ if (StringUtils.isBlank(masterUrl)) {
+ return null;
+ }
+ try {
+ URI uri = URI.create(masterUrl);
+ if (("ws".equalsIgnoreCase(uri.getScheme()) ||
"wss".equalsIgnoreCase(uri.getScheme())) &&
StringUtils.isNotBlank(uri.getHost())) {
+ return uri;
+ }
+ } catch (IllegalArgumentException ex) {
+ LOG.warn("Ignore invalid websocket master address [{}]",
masterUrl, ex);
+ return null;
+ }
+ LOG.warn("Ignore invalid websocket master address [{}]", masterUrl);
+ return null;
}
@Override
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 2ad99b3569..86df7d5af9 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
@@ -105,9 +105,9 @@ public final class ShenyuWebsocketClient extends
WebSocketClient {
private TimerTask timerTask;
- private String runningMode;
+ private volatile String runningMode;
- private String masterUrl;
+ private volatile String masterUrl;
private volatile boolean isConnectedToMaster;
@@ -265,6 +265,9 @@ public final class ShenyuWebsocketClient extends
WebSocketClient {
@Override
public void onOpen(final ServerHandshake serverHandshake) {
+ runningMode = null;
+ masterUrl = null;
+ isConnectedToMaster = false;
LOG.info("websocket connection server[{}] is opened, sending sync
msg", this.getURI().toString());
send(DataEventTypeEnum.RUNNING_MODE.name());
if (!alreadySync) {
@@ -294,11 +297,12 @@ 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)) {
- this.runningMode =
String.valueOf(jsonToMap.get(RunningModeConstants.RUNNING_MODE));
- if (!Objects.equals(RunningModeEnum.STANDALONE.name(),
runningMode)) {
+ 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));
}
+ this.runningMode = mode;
} else if (Objects.nonNull(initialSyncState)) {
initialSyncState.applyIncremental(() -> handleResult(result));
} else {
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
new file mode 100644
index 0000000000..922085d44e
--- /dev/null
+++
b/shenyu-sync-data-center/shenyu-sync-data-websocket/src/test/java/org/apache/shenyu/plugin/sync/data/websocket/WebsocketMasterRedirectTest.java
@@ -0,0 +1,310 @@
+/*
+ * 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.constant.Constants;
+import org.apache.shenyu.common.timer.Timer;
+import org.apache.shenyu.common.timer.WheelTimerFactory;
+import org.apache.shenyu.common.utils.GsonUtils;
+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.java_websocket.WebSocket;
+import org.java_websocket.handshake.ClientHandshake;
+import org.java_websocket.server.WebSocketServer;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+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 java.lang.reflect.Field;
+import java.net.InetSocketAddress;
+import java.net.URI;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.awaitility.Awaitility.await;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mockConstruction;
+import static org.mockito.Mockito.mockStatic;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * Exercises master discovery through real sockets and manually triggered
checks.
+ */
+class WebsocketMasterRedirectTest {
+
+ private final List<TestServer> servers = new ArrayList<>();
+
+ private MockedStatic<WheelTimerFactory> timers;
+
+ private WebsocketSyncDataService service;
+
+ private final CountDownLatch applied = new CountDownLatch(1);
+
+ @BeforeEach
+ void setUp() {
+ timers = mockStatic(WheelTimerFactory.class);
+
timers.when(WheelTimerFactory::newWheelTimer).thenReturn(mock(Timer.class));
+
timers.when(WheelTimerFactory::getSharedTimer).thenReturn(mock(Timer.class));
+ }
+
+ @AfterEach
+ void tearDown() throws InterruptedException {
+ try {
+ if (Objects.nonNull(service)) {
+ service.close();
+ }
+ for (TestServer server : servers) {
+ server.stop(1000);
+ }
+ } finally {
+ timers.close();
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void testRedirectsFollowerConnectionsToOneMaster(final boolean
multipleFollowers) throws Exception {
+ TestServer master = startServer(true);
+ TestServer follower = startServer(false);
+ follower.masterUrl = master.url();
+ final List<String> seeds = new
ArrayList<>(Collections.singletonList(follower.url()));
+ if (multipleFollowers) {
+ TestServer other = startServer(false);
+ other.masterUrl = master.url();
+ seeds.add(other.url());
+ }
+ createService(seeds);
+ awaitModes();
+
+ check();
+
+ assertTrue(applied.await(5, TimeUnit.SECONDS), "Configuration should
arrive from the advertised master");
+ await().atMost(Duration.ofSeconds(5)).until(() -> clients().size() ==
1 && clients().get(0).isConnectedToMaster());
+ assertEquals(URI.create(master.url()), clients().get(0).getURI());
+ assertEquals(1, master.opens.get());
+ assertTrue(follower.closed.await(5, TimeUnit.SECONDS));
+ assertEquals("test-token",
master.handshake.getFieldValue(Constants.X_SHENYU_SYNC_TOKEN));
+ assertEquals("https://gateway.example";,
master.handshake.getFieldValue("Origin"));
+ assertEquals("cluster-test",
master.handshake.getFieldValue(Constants.SHENYU_NAMESPACE_ID));
+ assertEquals("9195",
master.handshake.getFieldValue(Constants.CLIENT_PORT_NAME));
+ assertEquals(seeds, service.getWebsocketConfig().getUrls());
+ check();
+ assertEquals(1, master.opens.get());
+ assertEquals(1, follower.opens.get());
+ }
+
+ @Test
+ void testKeepsConnectionWhileRunningModeReplyIsPending() throws Exception {
+ TestServer server = startServer(true);
+ server.replyMode = false;
+ createService(Collections.singletonList(server.url()));
+ ShenyuWebsocketClient client = clients().get(0);
+
+ check();
+
+ assertEquals(1, clients().size());
+ assertSame(client, clients().get(0));
+ assertTrue(client.isOpen());
+ server.sendMode();
+ awaitModes();
+ check();
+ assertSame(client, clients().get(0));
+ }
+
+ @ParameterizedTest
+ @NullAndEmptySource
+ @ValueSource(strings = {"null", "/websocket", "http://admin/websocket";,
"ws://[invalid"})
+ void testInvalidMasterAddressFallsBackToSeeds(final String address) throws
Exception {
+ createService(Collections.emptyList());
+ ShenyuWebsocketClient follower = mock(ShenyuWebsocketClient.class);
+ when(follower.isOpen()).thenReturn(true);
+ when(follower.getRunningMode()).thenReturn("CLUSTER");
+ when(follower.getMasterUrl()).thenReturn(address);
+ clients().add(follower);
+ try (MockedConstruction<ShenyuWebsocketClient> constructed =
mockConstruction(ShenyuWebsocketClient.class)) {
+ check();
+ assertTrue(clients().isEmpty());
+ assertTrue(constructed.constructed().isEmpty());
+ verify(follower).nowClose();
+ }
+ }
+
+ @Test
+ void testReusesExistingMasterConnection() throws Exception {
+ createService(Collections.emptyList());
+ ShenyuWebsocketClient follower = mock(ShenyuWebsocketClient.class);
+ when(follower.isOpen()).thenReturn(true);
+ when(follower.getRunningMode()).thenReturn("CLUSTER");
+ when(follower.getMasterUrl()).thenReturn("ws://master:9095/websocket");
+ ShenyuWebsocketClient master = mock(ShenyuWebsocketClient.class);
+ when(master.isOpen()).thenReturn(true);
+ when(master.getRunningMode()).thenReturn("CLUSTER");
+ when(master.isConnectedToMaster()).thenReturn(true);
+
when(master.getURI()).thenReturn(URI.create("ws://master:9095/websocket"));
+ clients().add(follower);
+ clients().add(master);
+ try (MockedConstruction<ShenyuWebsocketClient> constructed =
mockConstruction(ShenyuWebsocketClient.class)) {
+ check();
+ assertEquals(Collections.singletonList(master), clients());
+ assertTrue(constructed.constructed().isEmpty());
+ }
+ }
+
+ @Test
+ void testClosedRedirectConnectionReturnsToConfiguredSeed() throws
Exception {
+ createService(Collections.emptyList());
+
service.getWebsocketConfig().setUrls(Collections.singletonList("ws://seed:9095/websocket"));
+ ShenyuWebsocketClient failedMaster = mock(ShenyuWebsocketClient.class);
+ clients().add(failedMaster);
+ check();
+ verify(failedMaster).nowClose();
+ try (MockedConstruction<ShenyuWebsocketClient> constructed =
mockConstruction(ShenyuWebsocketClient.class, (client, context) ->
+ assertEquals(URI.create("ws://seed:9095/websocket"),
context.arguments().get(0)))) {
+ check();
+ assertEquals(1, constructed.constructed().size());
+ }
+ }
+
+ private void createService(final List<String> seeds) {
+ WebsocketConfig config = new WebsocketConfig();
+ config.setUrls(seeds);
+ config.setToken("test-token");
+ 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);
+ }
+
+ private TestServer startServer(final boolean master) throws
InterruptedException {
+ TestServer server = new TestServer(master);
+ servers.add(server);
+ server.start();
+ assertTrue(server.started.await(5, TimeUnit.SECONDS));
+ return server;
+ }
+
+ private void awaitModes() {
+ await().atMost(Duration.ofSeconds(5)).until(() ->
clients().stream().allMatch(client ->
Objects.nonNull(client.getRunningMode())));
+ }
+
+ @SuppressWarnings("unchecked")
+ private List<ShenyuWebsocketClient> clients() throws
ReflectiveOperationException {
+ Field field =
WebsocketSyncDataService.class.getDeclaredField("clients");
+ field.setAccessible(true);
+ return (List<ShenyuWebsocketClient>) field.get(service);
+ }
+
+ private void check() throws ReflectiveOperationException {
+ java.lang.reflect.Method method =
WebsocketSyncDataService.class.getDeclaredMethod("masterCheck");
+ method.setAccessible(true);
+ method.invoke(service);
+ }
+
+ private static final class TestServer extends WebSocketServer {
+
+ private final boolean master;
+
+ private final CountDownLatch started = new CountDownLatch(1);
+
+ private final CountDownLatch closed = new CountDownLatch(1);
+
+ private final AtomicInteger opens = new AtomicInteger();
+
+ private volatile String masterUrl;
+
+ private volatile boolean replyMode = true;
+
+ private volatile WebSocket connection;
+
+ private volatile ClientHandshake handshake;
+
+ private TestServer(final boolean master) {
+ super(new InetSocketAddress("127.0.0.1", 0));
+ this.master = master;
+ }
+
+ private String url() {
+ return "ws://127.0.0.1:" + getPort() + "/websocket";
+ }
+
+ private void sendMode() {
+ connection.send(GsonUtils.getInstance().toJson(Map.of("eventType",
"RUNNING_MODE", "runningMode", "CLUSTER",
+ "isMaster", master, "masterUrl", master ? url() :
masterUrl)));
+ }
+
+ @Override
+ public void onOpen(final WebSocket socket, final ClientHandshake
clientHandshake) {
+ connection = socket;
+ handshake = clientHandshake;
+ opens.incrementAndGet();
+ }
+
+ @Override
+ public void onClose(final WebSocket socket, final int code, final
String reason, final boolean remote) {
+ closed.countDown();
+ }
+
+ @Override
+ public void onMessage(final WebSocket socket, final String message) {
+ if ("RUNNING_MODE".equals(message) && replyMode) {
+ sendMode();
+ } else if ("MYSELF".equals(message) && master) {
+
socket.send("{\"groupType\":\"PLUGIN\",\"eventType\":\"CREATE\",\"data\":[{\"id\":\"divide\",\"name\":\"divide\",\"enabled\":true}]}");
+ }
+ }
+
+ @Override
+ public void onError(final WebSocket socket, final Exception exception)
{
+ throw new IllegalStateException(exception);
+ }
+
+ @Override
+ public void onStart() {
+ started.countDown();
+ }
+ }
+}
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 deb8eeee9c..7c51d9f92b 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
@@ -118,6 +118,20 @@ public class ShenyuWebsocketClientTest {
verify(shenyuWebsocketClient).send(DataEventTypeEnum.MYSELF.name());
}
+ @Test
+ void testOnOpenClearsPreviousRunningMode() {
+ ShenyuWebsocketClient client = spy(shenyuWebsocketClient);
+ doNothing().when(client).send(anyString());
+
client.onMessage("{\"eventType\":\"RUNNING_MODE\",\"runningMode\":\"CLUSTER\",\"isMaster\":true,\"masterUrl\":\"ws://old-master/websocket\"}");
+ assertTrue(client.isConnectedToMaster());
+
+ client.onOpen(mock(ServerHandshake.class));
+
+ Assertions.assertNull(client.getRunningMode());
+ Assertions.assertNull(client.getMasterUrl());
+ assertFalse(client.isConnectedToMaster());
+ }
+
@Test
public void testOnMessage() {
doNothing().when(pluginDataSubscriber).onPluginRefresh(any());
