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 a32cff4850 fix(response): avoid resubscribing response bodies (#7087)
a32cff4850 is described below
commit a32cff4850afd74696ddb5bca58ce7b73d22a48e
Author: Liming Deng <[email protected]>
AuthorDate: Wed Sep 30 09:57:39 2026 +0800
fix(response): avoid resubscribing response bodies (#7087)
* fix(response): avoid resubscribing response bodies
* test(response): verify error cleanup paths
---
.../strategy/NettyClientMessageWriter.java | 9 ++--
.../response/strategy/WebClientMessageWriter.java | 17 ++------
.../strategy/NettyClientMessageWriterTest.java | 50 ++++++++++++++++++++++
.../strategy/WebClientMessageWriterTest.java | 48 ++++++++++++++++++++-
4 files changed, 103 insertions(+), 21 deletions(-)
diff --git
a/shenyu-plugin/shenyu-plugin-response/src/main/java/org/apache/shenyu/plugin/response/strategy/NettyClientMessageWriter.java
b/shenyu-plugin/shenyu-plugin-response/src/main/java/org/apache/shenyu/plugin/response/strategy/NettyClientMessageWriter.java
index c548a77166..65af21d99e 100644
---
a/shenyu-plugin/shenyu-plugin-response/src/main/java/org/apache/shenyu/plugin/response/strategy/NettyClientMessageWriter.java
+++
b/shenyu-plugin/shenyu-plugin-response/src/main/java/org/apache/shenyu/plugin/response/strategy/NettyClientMessageWriter.java
@@ -68,7 +68,8 @@ public class NettyClientMessageWriter implements
MessageWriter {
.inbound()
.receive()
.retain()
- .map(factory::wrap);
+ .map(factory::wrap)
+ .doOnDiscard(NettyDataBuffer.class,
DataBufferUtils::release);
MediaType contentType = response.getHeaders().getContentType();
Mono<Void> responseMono = isStreamingMediaType(contentType)
@@ -78,7 +79,7 @@ public class NettyClientMessageWriter implements
MessageWriter {
// watcher httpStatus
final Consumer<HttpStatusCode> consumer =
exchange.getAttribute(Constants.WATCHER_HTTP_STATUS);
Optional.ofNullable(consumer).ifPresent(c ->
c.accept(response.getStatusCode()));
- return responseMono.onErrorResume(error ->
releaseIfNotConsumed(body, error));
+ return responseMono.doOnError(error -> cleanup(exchange));
})).doOnCancel(() -> cleanup(exchange));
}
@@ -94,10 +95,6 @@ public class NettyClientMessageWriter implements
MessageWriter {
}
}
- private static <T> Mono<T> releaseIfNotConsumed(final
Flux<NettyDataBuffer> dataBufferDody, final Throwable ex) {
- return Objects.nonNull(dataBufferDody) ?
dataBufferDody.map(DataBufferUtils::release).then(Mono.error(ex)) :
Mono.error(ex);
- }
-
private boolean isStreamingMediaType(@Nullable final MediaType
contentType) {
return Objects.nonNull(contentType) &&
this.streamingMediaTypes.stream().anyMatch(contentType::isCompatibleWith);
}
diff --git
a/shenyu-plugin/shenyu-plugin-response/src/main/java/org/apache/shenyu/plugin/response/strategy/WebClientMessageWriter.java
b/shenyu-plugin/shenyu-plugin-response/src/main/java/org/apache/shenyu/plugin/response/strategy/WebClientMessageWriter.java
index c3f9de6b8a..9f4c84b06c 100644
---
a/shenyu-plugin/shenyu-plugin-response/src/main/java/org/apache/shenyu/plugin/response/strategy/WebClientMessageWriter.java
+++
b/shenyu-plugin/shenyu-plugin-response/src/main/java/org/apache/shenyu/plugin/response/strategy/WebClientMessageWriter.java
@@ -80,9 +80,9 @@ public class WebClientMessageWriter implements MessageWriter {
Mono<Void> responseMono;
if (Objects.nonNull(fluxResponseEntity.getBody())) {
- responseMono =
exchange.getResponse().writeWith(fluxResponseEntity.getBody())
- .onErrorResume(error ->
releaseIfNotConsumed(fluxResponseEntity.getBody(), error))
- .doOnCancel(() -> clean(exchange));
+ Flux<DataBuffer> body = fluxResponseEntity.getBody()
+ .doOnDiscard(DataBuffer.class,
DataBufferUtils::release);
+ responseMono = exchange.getResponse().writeWith(body);
} else {
responseMono = exchange.getResponse().writeWith(Mono.empty());
}
@@ -121,17 +121,6 @@ public class WebClientMessageWriter implements
MessageWriter {
response.getHeaders().putAll(httpHeaders);
}
- private static <T> Mono<T> releaseIfNotConsumed(final Flux<DataBuffer>
dataBufferDody, final Throwable ex) {
- return
dataBufferDody.map(DataBufferUtils::release).then(Mono.error(ex));
- }
-
- private void clean(final ServerWebExchange exchange) {
- ResponseEntity<Flux<DataBuffer>> fluxResponseEntity =
exchange.getAttribute(Constants.CLIENT_RESPONSE_ATTR);
- if (Objects.nonNull(fluxResponseEntity) &&
Objects.nonNull(fluxResponseEntity.getBody())) {
-
fluxResponseEntity.getBody().map(DataBufferUtils::release).subscribe();
- }
- }
-
static {
// https://www.iana.org/assignments/media-types/media-types.xhtml
//
https://developer.mozilla.org/en-US/docs/Web/HTTP/Basics_of_HTTP/MIME_types
diff --git
a/shenyu-plugin/shenyu-plugin-response/src/test/java/org/apache/shenyu/plugin/response/strategy/NettyClientMessageWriterTest.java
b/shenyu-plugin/shenyu-plugin-response/src/test/java/org/apache/shenyu/plugin/response/strategy/NettyClientMessageWriterTest.java
index 7c5baceb60..9e14789aed 100644
---
a/shenyu-plugin/shenyu-plugin-response/src/test/java/org/apache/shenyu/plugin/response/strategy/NettyClientMessageWriterTest.java
+++
b/shenyu-plugin/shenyu-plugin-response/src/test/java/org/apache/shenyu/plugin/response/strategy/NettyClientMessageWriterTest.java
@@ -18,6 +18,7 @@
package org.apache.shenyu.plugin.response.strategy;
import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.PooledByteBufAllocator;
import org.apache.shenyu.common.constant.Constants;
import org.apache.shenyu.plugin.api.ShenyuPluginChain;
import org.apache.shenyu.plugin.api.result.ShenyuResult;
@@ -25,6 +26,8 @@ import org.apache.shenyu.plugin.api.utils.SpringBeanUtils;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.BeforeEach;
import org.springframework.context.ConfigurableApplicationContext;
+import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.core.io.buffer.DataBufferUtils;
import org.springframework.core.io.buffer.NettyDataBuffer;
import org.springframework.core.io.buffer.NettyDataBufferFactory;
import org.springframework.http.HttpHeaders;
@@ -36,13 +39,20 @@ import org.springframework.web.server.ServerWebExchange;
import reactor.core.CoreSubscriber;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
+import reactor.core.scheduler.Schedulers;
import reactor.netty.ByteBufFlux;
import reactor.netty.Connection;
import reactor.netty.NettyInbound;
import reactor.test.StepVerifier;
+import org.reactivestreams.Publisher;
+
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
/**
@@ -93,4 +103,44 @@ public class NettyClientMessageWriterTest {
StepVerifier.create(nettyClientMessageWriter.writeWith(exchange,
chain)).expectSubscription().verifyError();
}
+
+ @Test
+ public void testWriteErrorReleasesBuffersAndDisposesConnection() {
+ RuntimeException expected = new RuntimeException("write failed");
+ AtomicInteger subscriptions = new AtomicInteger();
+ NettyDataBufferFactory factory = new
NettyDataBufferFactory(PooledByteBufAllocator.DEFAULT);
+ NettyDataBuffer consumed = factory.allocateBuffer(1);
+ NettyDataBuffer discarded = factory.allocateBuffer(1);
+ Flux<NettyDataBuffer> body = Flux.defer(() -> {
+ subscriptions.incrementAndGet();
+ return Flux.just(consumed,
discarded).publishOn(Schedulers.immediate(), 2);
+ });
+ ServerWebExchange exchange = mock(ServerWebExchange.class);
+ ServerHttpResponse response = mock(ServerHttpResponse.class);
+ Connection connection = mock(Connection.class);
+ NettyInbound inbound = mock(NettyInbound.class);
+ ByteBufFlux received = mock(ByteBufFlux.class);
+ ByteBufFlux retained = mock(ByteBufFlux.class);
+ when(exchange.getResponse()).thenReturn(response);
+ when(response.getHeaders()).thenReturn(new HttpHeaders());
+ when(response.bufferFactory()).thenReturn(factory);
+
when(exchange.getAttribute(Constants.CLIENT_RESPONSE_CONN_ATTR)).thenReturn(connection);
+ when(connection.inbound()).thenReturn(inbound);
+ when(inbound.receive()).thenReturn(received);
+ when(received.retain()).thenReturn(retained);
+ when(retained.<NettyDataBuffer>map(any())).thenReturn(body);
+ when(response.writeWith(any())).thenAnswer(invocation ->
Flux.from((Publisher<DataBuffer>) invocation.getArgument(0))
+ .take(1)
+ .doOnNext(DataBufferUtils::release)
+ .then(Mono.error(expected)));
+
+ StepVerifier.create(nettyClientMessageWriter.writeWith(exchange,
chain))
+ .expectErrorMatches(error -> error == expected)
+ .verify();
+
+ assertEquals(1, subscriptions.get());
+ assertEquals(0, consumed.getNativeBuffer().refCnt());
+ assertEquals(0, discarded.getNativeBuffer().refCnt());
+ verify(connection).dispose();
+ }
}
diff --git
a/shenyu-plugin/shenyu-plugin-response/src/test/java/org/apache/shenyu/plugin/response/strategy/WebClientMessageWriterTest.java
b/shenyu-plugin/shenyu-plugin-response/src/test/java/org/apache/shenyu/plugin/response/strategy/WebClientMessageWriterTest.java
index a065cfc3a3..ad20e343bd 100644
---
a/shenyu-plugin/shenyu-plugin-response/src/test/java/org/apache/shenyu/plugin/response/strategy/WebClientMessageWriterTest.java
+++
b/shenyu-plugin/shenyu-plugin-response/src/test/java/org/apache/shenyu/plugin/response/strategy/WebClientMessageWriterTest.java
@@ -17,6 +17,7 @@
package org.apache.shenyu.plugin.response.strategy;
+import io.netty.buffer.PooledByteBufAllocator;
import org.apache.shenyu.common.constant.Constants;
import org.apache.shenyu.plugin.api.ShenyuPluginChain;
import org.apache.shenyu.plugin.api.context.ShenyuContext;
@@ -29,17 +30,30 @@ import org.mockito.junit.jupiter.MockitoExtension;
import org.mockito.junit.jupiter.MockitoSettings;
import org.mockito.quality.Strictness;
import org.springframework.context.ConfigurableApplicationContext;
+import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.core.io.buffer.DataBufferUtils;
+import org.springframework.core.io.buffer.NettyDataBuffer;
+import org.springframework.core.io.buffer.NettyDataBufferFactory;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.http.codec.ServerCodecConfigurer;
import org.springframework.http.codec.support.DefaultServerCodecConfigurer;
+import org.springframework.http.server.reactive.ServerHttpResponse;
import org.springframework.mock.http.server.reactive.MockServerHttpRequest;
import org.springframework.mock.web.server.MockServerWebExchange;
import org.springframework.web.server.ServerWebExchange;
+import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
+import reactor.core.scheduler.Schedulers;
import reactor.test.StepVerifier;
+import org.reactivestreams.Publisher;
+
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.reset;
import static org.mockito.Mockito.when;
@@ -102,10 +116,42 @@ public class WebClientMessageWriterTest {
StepVerifier.create(monoGatewayTimeout).expectSubscription().verifyComplete();
}
+ @Test
+ public void testWriteErrorDoesNotResubscribeResponseBody() {
+ RuntimeException expected = new RuntimeException("write failed");
+ AtomicInteger subscriptions = new AtomicInteger();
+ NettyDataBufferFactory factory = new
NettyDataBufferFactory(PooledByteBufAllocator.DEFAULT);
+ NettyDataBuffer consumed = factory.allocateBuffer(1);
+ NettyDataBuffer discarded = factory.allocateBuffer(1);
+ Flux<DataBuffer> body = Flux.defer(() -> {
+ subscriptions.incrementAndGet();
+ return Flux.<DataBuffer>just(consumed,
discarded).publishOn(Schedulers.immediate(), 2);
+ });
+ ResponseEntity<Flux<DataBuffer>> clientResponse =
ResponseEntity.ok(body);
+ ServerWebExchange exchange = mock(ServerWebExchange.class);
+ ServerHttpResponse response = mock(ServerHttpResponse.class);
+ when(exchange.getResponse()).thenReturn(response);
+ when(response.getHeaders()).thenReturn(new HttpHeaders());
+
when(exchange.getAttribute(Constants.CLIENT_RESPONSE_ATTR)).thenReturn(clientResponse);
+ when(response.writeWith(any())).thenAnswer(invocation ->
Flux.from((Publisher<DataBuffer>) invocation.getArgument(0))
+ .take(1)
+ .doOnNext(DataBufferUtils::release)
+ .then(Mono.error(expected)));
+ when(chain.execute(exchange)).thenReturn(Mono.empty());
+
+ StepVerifier.create(webClientMessageWriter.writeWith(exchange, chain))
+ .expectErrorMatches(error -> error == expected)
+ .verify();
+
+ assertEquals(1, subscriptions.get());
+ assertEquals(0, consumed.getNativeBuffer().refCnt());
+ assertEquals(0, discarded.getNativeBuffer().refCnt());
+ }
+
private ServerWebExchange generateServerWebExchange(final boolean
haveResponse) {
ResponseEntity mockResponse = mock(ResponseEntity.class);
when(mockResponse.getHeaders()).thenReturn(mock(HttpHeaders.class));
- when(mockResponse.getBody()).thenReturn(Mono.empty());
+ when(mockResponse.getBody()).thenReturn(Flux.empty());
ServerWebExchange exchange = MockServerWebExchange
.from(MockServerHttpRequest.get("/test").build());
