From 89d56b1fa648a1e6b3d17e44c14156a437187966 Mon Sep 17 00:00:00 2001 From: rstoyanchev Date: Mon, 28 Oct 2024 09:11:27 +0000 Subject: [PATCH] Streaming ReactorClientHttpResponse Closes gh-33781 --- .../client/ReactorClientHttpResponse.java | 21 ++++++++++++------- 1 file changed, 13 insertions(+), 8 deletions(-) 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