diff --git a/spring-web/src/main/java/org/springframework/http/client/ReactorClientHttpResponse.java b/spring-web/src/main/java/org/springframework/http/client/ReactorClientHttpResponse.java index 22c242363a..593ace979c 100644 --- a/spring-web/src/main/java/org/springframework/http/client/ReactorClientHttpResponse.java +++ b/spring-web/src/main/java/org/springframework/http/client/ReactorClientHttpResponse.java @@ -20,6 +20,8 @@ import java.io.IOException; import java.io.InputStream; import java.time.Duration; +import io.netty.buffer.ByteBuf; +import org.reactivestreams.FlowAdapters; import reactor.netty.Connection; import reactor.netty.http.client.HttpClientResponse; @@ -79,19 +81,22 @@ final class ReactorClientHttpResponse implements ClientHttpResponse { if (body != null) { return body; } - try { - body = this.connection.inbound().receive().aggregate().asInputStream().block(this.readTimeout); + SubscriberInputStream is = new SubscriberInputStream<>( + byteBuf -> { + byte[] bytes = new byte[byteBuf.readableBytes()]; + byteBuf.readBytes(bytes); + byteBuf.release(); + return bytes; + }, + ByteBuf::release, 16); + this.connection.inbound().receive().retain().subscribe(FlowAdapters.toSubscriber(is)); + this.body = is; + return is; } catch (RuntimeException ex) { throw ReactorClientHttpRequest.convertException(ex); } - - if (body == null) { - body = InputStream.nullInputStream(); - } - this.body = body; - return body; } @Override