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 70fba09c3f fix(web): scope chain timing and execution to each
subscription (#7793)
70fba09c3f is described below
commit 70fba09c3f817786b1534708cc77785d7cf9c211
Author: redyyang <[email protected]>
AuthorDate: Thu Oct 22 10:26:57 2026 +0800
fix(web): scope chain timing and execution to each subscription (#7793)
---
.../org/apache/shenyu/plugin/api/ShenyuPlugin.java | 9 +
.../shenyu/web/handler/ShenyuWebHandler.java | 32 ++--
.../web/handler/ShenyuWebHandlerLifecycleTest.java | 210 +++++++++++++++++++++
.../shenyu/web/handler/ShenyuWebHandlerTest.java | 6 +-
4 files changed, 240 insertions(+), 17 deletions(-)
diff --git
a/shenyu-plugin/shenyu-plugin-api/src/main/java/org/apache/shenyu/plugin/api/ShenyuPlugin.java
b/shenyu-plugin/shenyu-plugin-api/src/main/java/org/apache/shenyu/plugin/api/ShenyuPlugin.java
index 008f60bd88..fe770db851 100644
---
a/shenyu-plugin/shenyu-plugin-api/src/main/java/org/apache/shenyu/plugin/api/ShenyuPlugin.java
+++
b/shenyu-plugin/shenyu-plugin-api/src/main/java/org/apache/shenyu/plugin/api/ShenyuPlugin.java
@@ -136,6 +136,10 @@ public interface ShenyuPlugin {
/**
* Plugin before operation.
*
+ * <p>The gateway invokes this immediately before calling {@link
#execute}, during chain
+ * subscription. Together with {@link #after}, it brackets construction of
the plugin's
+ * publisher, not its asynchronous completion.</p>
+ *
* @param exchange context
*/
default void before(ServerWebExchange exchange) {
@@ -145,6 +149,11 @@ public interface ShenyuPlugin {
/**
* Plugin after operation.
*
+ * <p>The gateway invokes this when {@link #execute} returns or throws,
before subscribing
+ * to the returned publisher. These construction-time semantics are
retained for compatibility
+ * with plugin overrides, including WASM callbacks. The gateway chain
timing separately measures
+ * the entire subscription through its terminal signal.</p>
+ *
* @param exchange context
*/
default void after(ServerWebExchange exchange) {
diff --git
a/shenyu-web/src/main/java/org/apache/shenyu/web/handler/ShenyuWebHandler.java
b/shenyu-web/src/main/java/org/apache/shenyu/web/handler/ShenyuWebHandler.java
index 43dc50a19a..12fefc30ca 100644
---
a/shenyu-web/src/main/java/org/apache/shenyu/web/handler/ShenyuWebHandler.java
+++
b/shenyu-web/src/main/java/org/apache/shenyu/web/handler/ShenyuWebHandler.java
@@ -46,6 +46,7 @@ import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.Set;
+import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
/**
@@ -94,12 +95,12 @@ public final class ShenyuWebHandler implements WebHandler,
ApplicationListener<P
}
/**
- * Chain before operation.
+ * Start monotonic chain timing for a subscription, before any scheduler
queueing.
*
* @param exchange context
*/
public void before(final ServerWebExchange exchange) {
- exchange.getAttributes().put(Constants.CHAIN_START_TIME,
System.currentTimeMillis());
+ exchange.getAttributes().put(Constants.CHAIN_START_TIME,
System.nanoTime());
}
/**
@@ -108,8 +109,8 @@ public final class ShenyuWebHandler implements WebHandler,
ApplicationListener<P
* <p>It runs with the terminal signal of the request, not while the chain
is assembled, so the
* reported cost is the duration of the request and the trace id of the
gateway span is already
* populated. A long lived response, for example a streamed one, reports
its timing once the
- * stream has ended. The operation is idempotent: a request which is
subscribed again does not
- * report a second time.</p>
+ * stream has ended. Each subscription starts its own timing; repeated
cleanup of the same
+ * subscription is ignored. The cost is reported in milliseconds.</p>
*
* @param exchange context
*/
@@ -119,27 +120,30 @@ public final class ShenyuWebHandler implements
WebHandler, ApplicationListener<P
if (Objects.isNull(startTime)) {
return;
}
- long cost = System.currentTimeMillis() - (long) startTime;
+ long cost = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - (long)
startTime);
LOG.debug("shenyu chain handle uri:{}, traceId:{}, cost:{}",
exchange.getRequest().getPath(), TraceIdUtils.traceId(exchange), cost);
}
/**
* Handle the web server exchange.
*
+ * <p>Timing includes scheduler queueing and asynchronous execution until
completion, failure
+ * or cancellation. Creating the publisher does not start timing. Each
subscription creates
+ * a fresh chain, so sequential subscriptions execute all plugins
again.</p>
+ *
* @param exchange the current server exchange
* @return {@code Mono<Void>} to indicate when request handling is complete
*/
@Override
public Mono<Void> handle(@NonNull final ServerWebExchange exchange) {
- before(exchange);
- Mono<Void> execute = new
DefaultShenyuPluginChain(plugins).execute(exchange);
- if (scheduled) {
- execute = execute.subscribeOn(scheduler);
- }
- // the chain is assembled here but only runs once it is subscribed,
and the tracing span is
- // created at that point as well, so the timing and the trace id are
read with the terminal
- // signal of the request instead of while the chain is assembled.
- return execute.doFinally(signalType -> after(exchange));
+ return Mono.defer(() -> {
+ before(exchange);
+ Mono<Void> execute = new
DefaultShenyuPluginChain(plugins).execute(exchange);
+ if (scheduled) {
+ execute = execute.subscribeOn(scheduler);
+ }
+ return execute.doFinally(signalType -> after(exchange));
+ });
}
/**
diff --git
a/shenyu-web/src/test/java/org/apache/shenyu/web/handler/ShenyuWebHandlerLifecycleTest.java
b/shenyu-web/src/test/java/org/apache/shenyu/web/handler/ShenyuWebHandlerLifecycleTest.java
new file mode 100644
index 0000000000..5d16e21313
--- /dev/null
+++
b/shenyu-web/src/test/java/org/apache/shenyu/web/handler/ShenyuWebHandlerLifecycleTest.java
@@ -0,0 +1,210 @@
+/*
+ * 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.web.handler;
+
+import org.apache.shenyu.common.config.ShenyuConfig;
+import org.apache.shenyu.common.constant.Constants;
+import org.apache.shenyu.plugin.api.ShenyuPlugin;
+import org.apache.shenyu.web.loader.ShenyuLoaderService;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+import org.springframework.mock.http.server.reactive.MockServerHttpRequest;
+import org.springframework.mock.web.server.MockServerWebExchange;
+import org.springframework.test.util.ReflectionTestUtils;
+import reactor.core.Disposable;
+import reactor.core.Disposables;
+import reactor.core.publisher.Mono;
+import reactor.core.publisher.Sinks;
+import reactor.core.scheduler.Scheduler;
+import reactor.test.StepVerifier;
+import reactor.test.scheduler.VirtualTimeScheduler;
+
+import java.time.Duration;
+import java.util.Collections;
+import java.util.Objects;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * Tests subscription and terminal-signal lifecycle with controlled publishers
and scheduling.
+ */
+class ShenyuWebHandlerLifecycleTest {
+
+ private final ShenyuPlugin plugin = mock(ShenyuPlugin.class);
+
+ private final MockServerWebExchange exchange =
MockServerWebExchange.from(MockServerHttpRequest.get("/lifecycle"));
+
+ private final VirtualTimeScheduler scheduler =
VirtualTimeScheduler.create();
+
+ @AfterEach
+ void tearDown() {
+ scheduler.dispose();
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void testCompletion(final boolean scheduled) {
+ verifyTerminal(scheduled, null, false);
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void testAsynchronousFailure(final boolean scheduled) {
+ verifyTerminal(scheduled, new IllegalStateException("upstream
failed"), false);
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void testCancellation(final boolean scheduled) {
+ verifyTerminal(scheduled, null, true);
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void testSynchronousPluginFailure(final boolean scheduled) {
+ ShenyuWebHandler handler = handler(scheduled);
+ IllegalStateException failure = new IllegalStateException("plugin
failed");
+ when(plugin.execute(any(), any())).thenThrow(failure);
+
+ StepVerifier.create(handler.handle(exchange))
+ .then(() -> scheduler.advanceTimeBy(Duration.ZERO))
+ .expectErrorSatisfies(error -> assertSame(failure, error))
+ .verify(Duration.ofSeconds(5));
+
+ verify(handler).before(exchange);
+ verify(handler).after(exchange);
+ verify(plugin).after(exchange);
+ assertNull(exchange.getAttribute(Constants.CHAIN_START_TIME));
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void testTimingRemainsActiveUntilDelayedCompletion(final boolean
scheduled) {
+ ShenyuWebHandler handler = handler(scheduled);
+ when(plugin.execute(any(),
any())).thenReturn(Mono.delay(Duration.ofSeconds(5), scheduler).then());
+ final Mono<Void> result = handler.handle(exchange);
+ verify(handler, never()).before(exchange);
+ verify(handler, never()).after(exchange);
+ assertNull(exchange.getAttribute(Constants.CHAIN_START_TIME));
+
+ StepVerifier.create(result)
+ .then(() -> scheduler.advanceTimeBy(Duration.ofSeconds(4)))
+ .then(() -> {
+
assertNotNull(exchange.getAttribute(Constants.CHAIN_START_TIME));
+ verify(handler, never()).after(exchange);
+ })
+ .then(() -> scheduler.advanceTimeBy(Duration.ofSeconds(1)))
+ .expectComplete()
+ .verify(Duration.ofSeconds(5));
+
+ verify(handler).before(exchange);
+ verify(handler).after(exchange);
+ assertNull(exchange.getAttribute(Constants.CHAIN_START_TIME));
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void testEachSubscriptionExecutesAFreshChain(final boolean scheduled) {
+ ShenyuWebHandler handler = handler(scheduled);
+ when(plugin.execute(any(), any())).thenReturn(Mono.empty());
+ final Mono<Void> result = handler.handle(exchange);
+ for (int subscription = 0; subscription < 2; subscription++) {
+ StepVerifier.create(result)
+ .then(() -> scheduler.advanceTimeBy(Duration.ZERO))
+ .expectComplete().verify(Duration.ofSeconds(5));
+ assertNull(exchange.getAttribute(Constants.CHAIN_START_TIME));
+ }
+ verify(plugin, times(2)).execute(any(), any());
+ verify(handler, times(2)).before(exchange);
+ verify(handler, times(2)).after(exchange);
+ }
+
+ @Test
+ void testCancellationBeforeScheduledExecution() {
+ ShenyuWebHandler handler = handler(true);
+ Scheduler queuedScheduler = mock(Scheduler.class);
+ Scheduler.Worker worker = mock(Scheduler.Worker.class);
+ when(queuedScheduler.createWorker()).thenReturn(worker);
+
when(worker.schedule(any(Runnable.class))).thenReturn(Disposables.single());
+ ReflectionTestUtils.setField(handler, "scheduler", queuedScheduler);
+ Disposable subscription = handler.handle(exchange).subscribe();
+ assertNotNull(exchange.getAttribute(Constants.CHAIN_START_TIME));
+ verify(plugin, never()).execute(any(), any());
+
+ subscription.dispose();
+ scheduler.advanceTimeBy(Duration.ZERO);
+
+ verify(plugin, never()).execute(any(), any());
+ verify(handler).before(exchange);
+ verify(handler).after(exchange);
+ verify(worker).dispose();
+ assertNull(exchange.getAttribute(Constants.CHAIN_START_TIME));
+ }
+
+ private void verifyTerminal(final boolean scheduled, final Throwable
failure, final boolean cancel) {
+ ShenyuWebHandler handler = handler(scheduled);
+ Sinks.Empty<Void> pending = Sinks.empty();
+ when(plugin.execute(any(), any())).thenReturn(pending.asMono());
+ final Mono<Void> result = handler.handle(exchange);
+ verify(handler, never()).before(exchange);
+ verify(handler, never()).after(exchange);
+ assertNull(exchange.getAttribute(Constants.CHAIN_START_TIME));
+ AtomicReference<Throwable> observedFailure = new AtomicReference<>();
+ Disposable subscription = result.subscribe(ignored -> { },
observedFailure::set);
+ scheduler.advanceTimeBy(Duration.ZERO);
+ verify(handler).before(exchange);
+ verify(handler, never()).after(exchange);
+ assertNotNull(exchange.getAttribute(Constants.CHAIN_START_TIME));
+ // Plugin hooks still bracket publisher construction, including WASM
callbacks.
+ verify(plugin).before(exchange);
+ verify(plugin).after(exchange);
+ if (cancel) {
+ subscription.dispose();
+ } else if (Objects.isNull(failure)) {
+ pending.tryEmitEmpty();
+ } else {
+ pending.tryEmitError(failure);
+ }
+ assertSame(failure, observedFailure.get());
+ verify(handler).after(exchange);
+ assertNull(exchange.getAttribute(Constants.CHAIN_START_TIME));
+ subscription.dispose();
+ verify(handler).after(exchange);
+ }
+
+ private ShenyuWebHandler handler(final boolean scheduled) {
+ ShenyuConfig config = new ShenyuConfig();
+ config.getScheduler().setEnabled(scheduled);
+ config.getScheduler().setType("elastic");
+ ShenyuWebHandler handler = spy(new
ShenyuWebHandler(Collections.singletonList(plugin),
mock(ShenyuLoaderService.class), config));
+ ReflectionTestUtils.setField(handler, "scheduler", scheduler);
+ return handler;
+ }
+}
diff --git
a/shenyu-web/src/test/java/org/apache/shenyu/web/handler/ShenyuWebHandlerTest.java
b/shenyu-web/src/test/java/org/apache/shenyu/web/handler/ShenyuWebHandlerTest.java
index 757084d9dd..502223a629 100644
---
a/shenyu-web/src/test/java/org/apache/shenyu/web/handler/ShenyuWebHandlerTest.java
+++
b/shenyu-web/src/test/java/org/apache/shenyu/web/handler/ShenyuWebHandlerTest.java
@@ -108,7 +108,7 @@ public final class ShenyuWebHandlerTest {
Mono<Void> handle = shenyuWebHandler.handle(exchange);
// the chain only runs once it is subscribed, the tracing span is
created at that point, so
// the chain timing is not read while the chain is assembled.
-
assertNotNull(exchange.getAttributes().get(Constants.CHAIN_START_TIME));
+ assertNull(exchange.getAttributes().get(Constants.CHAIN_START_TIME));
StepVerifier.create(handle).expectSubscription().verifyComplete();
assertNull(exchange.getAttributes().get(Constants.CHAIN_START_TIME));
}
@@ -147,7 +147,7 @@ public final class ShenyuWebHandlerTest {
}
@Test
- public void handleIsIdempotentWhenTheRequestIsSubscribedTwice() {
+ public void handleReportsEachSubscription() {
final ServerWebExchange exchange =
MockServerWebExchange.from(MockServerHttpRequest.get("localhost")
.remoteAddress(new InetSocketAddress(8090))
.build());
@@ -170,7 +170,7 @@ public final class ShenyuWebHandlerTest {
.map(ILoggingEvent::getFormattedMessage)
.filter(message -> message.startsWith("shenyu chain handle"))
.count();
- assertEquals(1, reported, "a request which is subscribed again must
not be reported again");
+ assertEquals(2, reported, "each subscription must have its own chain
timing");
}
@Test
