diff --git a/spring-core/src/main/java/org/springframework/core/io/buffer/DataBufferUtils.java b/spring-core/src/main/java/org/springframework/core/io/buffer/DataBufferUtils.java index 79bb1ce4b9..99656b4093 100644 --- a/spring-core/src/main/java/org/springframework/core/io/buffer/DataBufferUtils.java +++ b/spring-core/src/main/java/org/springframework/core/io/buffer/DataBufferUtils.java @@ -68,8 +68,6 @@ public abstract class DataBufferUtils { private static final Consumer RELEASE_CONSUMER = DataBufferUtils::release; - private static final int DEFAULT_CHUNK_SIZE = 1024; - //--------------------------------------------------------------------- // Reading @@ -442,48 +440,17 @@ public abstract class DataBufferUtils { public static Publisher outputStreamPublisher(Consumer outputStreamConsumer, DataBufferFactory bufferFactory, Executor executor) { - return outputStreamPublisher(outputStreamConsumer, bufferFactory, executor, DEFAULT_CHUNK_SIZE); + return new OutputStreamPublisher(outputStreamConsumer, bufferFactory, executor, null); } /** - * Creates a new {@code Publisher} based on bytes written to a - * {@code OutputStream}. - *
    - *
  • The parameter {@code outputStreamConsumer} is invoked once per - * subscription of the returned {@code Publisher}, when the first - * item is - * {@linkplain Subscription#request(long) requested}.
    • - *
  • {@link OutputStream#write(byte[], int, int) OutputStream.write()} - * invocations made by {@code outputStreamHandler} are buffered until they - * reach or exceed {@code chunkSize}, or when the stream is - * {@linkplain OutputStream#flush() flushed} and then result in a - * {@linkplain Subscriber#onNext(Object) published} item - * if there is {@linkplain Subscription#request(long) demand}.
    • - *
  • If there is no demand, {@code OutputStream.write()} will block - * until there is.
    • - *
  • If the subscription is {@linkplain Subscription#cancel() cancelled}, - * {@code OutputStream.write()} will throw a {@code IOException}.
    • - *
  • The subscription is - * {@linkplain Subscriber#onComplete() completed} when - * {@code outputStreamHandler} completes.
    • - *
  • Any exceptions thrown from {@code outputStreamHandler} will - * be dispatched to the {@linkplain Subscriber#onError(Throwable) Subscriber}. - *
- * @param outputStreamConsumer invoked when the first buffer is requested - * @param executor used to invoke the {@code outputStreamHandler} - * @param chunkSize minimum size of the buffer produced by the publisher - * @return a {@code Publisher} based on bytes written by - * {@code outputStreamHandler} + * Variant of {@link #outputStreamPublisher(Consumer, DataBufferFactory, Executor)} + * providing control over the chunk sizes to be produced by the publisher. * @since 6.1 */ public static Publisher outputStreamPublisher(Consumer outputStreamConsumer, DataBufferFactory bufferFactory, Executor executor, int chunkSize) { - Assert.notNull(outputStreamConsumer, "OutputStreamConsumer must not be null"); - Assert.notNull(bufferFactory, "BufferFactory must not be null"); - Assert.notNull(executor, "Executor must not be null"); - Assert.isTrue(chunkSize > 0, "Chunk size must be > 0"); - return new OutputStreamPublisher(outputStreamConsumer, bufferFactory, executor, chunkSize); } diff --git a/spring-core/src/main/java/org/springframework/core/io/buffer/OutputStreamPublisher.java b/spring-core/src/main/java/org/springframework/core/io/buffer/OutputStreamPublisher.java index 8e77f26eaa..d8ef636a95 100644 --- a/spring-core/src/main/java/org/springframework/core/io/buffer/OutputStreamPublisher.java +++ b/spring-core/src/main/java/org/springframework/core/io/buffer/OutputStreamPublisher.java @@ -31,10 +31,10 @@ import org.reactivestreams.Subscriber; import org.reactivestreams.Subscription; import org.springframework.lang.Nullable; +import org.springframework.util.Assert; /** - * Bridges between {@link OutputStream} and - * {@link Publisher Publisher<DataBuffer>}. + * Bridges between {@link OutputStream} and {@link Publisher Publisher<DataBuffer>}. * *

Note that this class has a near duplicate in * {@link org.springframework.http.client.OutputStreamPublisher}. @@ -45,6 +45,9 @@ import org.springframework.lang.Nullable; */ final class OutputStreamPublisher implements Publisher { + private static final int DEFAULT_CHUNK_SIZE = 1024; + + private final Consumer outputStreamConsumer; private final DataBufferFactory bufferFactory; @@ -54,14 +57,26 @@ final class OutputStreamPublisher implements Publisher { private final int chunkSize; + /** + * Create an instance. + * @param outputStreamConsumer invoked when the first buffer is requested + * @param bufferFactory to create data buffers with + * @param executor used to invoke the {@code outputStreamHandler} + * @param chunkSize the chunk sizes to be produced by the publisher + */ OutputStreamPublisher( Consumer outputStreamConsumer, DataBufferFactory bufferFactory, - Executor executor, int chunkSize) { + Executor executor, @Nullable Integer chunkSize) { + + Assert.notNull(outputStreamConsumer, "OutputStreamConsumer must not be null"); + Assert.notNull(bufferFactory, "BufferFactory must not be null"); + Assert.notNull(executor, "Executor must not be null"); + Assert.isTrue(chunkSize == null || chunkSize > 0, "ChunkSize must be larger than 0"); this.outputStreamConsumer = outputStreamConsumer; this.bufferFactory = bufferFactory; this.executor = executor; - this.chunkSize = chunkSize; + this.chunkSize = (chunkSize != null ? chunkSize : DEFAULT_CHUNK_SIZE); } diff --git a/spring-web/src/main/java/org/springframework/http/client/JdkClientHttpRequest.java b/spring-web/src/main/java/org/springframework/http/client/JdkClientHttpRequest.java index ae302bbf3d..aac03a1903 100644 --- a/spring-web/src/main/java/org/springframework/http/client/JdkClientHttpRequest.java +++ b/spring-web/src/main/java/org/springframework/http/client/JdkClientHttpRequest.java @@ -153,19 +153,18 @@ class JdkClientHttpRequest extends AbstractStreamingClientHttpRequest { private HttpRequest.BodyPublisher bodyPublisher(HttpHeaders headers, @Nullable Body body) { if (body != null) { - Flow.Publisher outputStreamPublisher = OutputStreamPublisher.create( - outputStream -> body.writeTo(StreamUtils.nonClosing(outputStream)), - BYTE_MAPPER, this.executor); + Flow.Publisher publisher = new OutputStreamPublisher<>( + os -> body.writeTo(StreamUtils.nonClosing(os)), BYTE_MAPPER, this.executor, null); long contentLength = headers.getContentLength(); if (contentLength > 0) { - return HttpRequest.BodyPublishers.fromPublisher(outputStreamPublisher, contentLength); + return HttpRequest.BodyPublishers.fromPublisher(publisher, contentLength); } else if (contentLength == 0) { return HttpRequest.BodyPublishers.noBody(); } else { - return HttpRequest.BodyPublishers.fromPublisher(outputStreamPublisher); + return HttpRequest.BodyPublishers.fromPublisher(publisher); } } else { diff --git a/spring-web/src/main/java/org/springframework/http/client/OutputStreamPublisher.java b/spring-web/src/main/java/org/springframework/http/client/OutputStreamPublisher.java index 6bfde107c9..023745db18 100644 --- a/spring-web/src/main/java/org/springframework/http/client/OutputStreamPublisher.java +++ b/spring-web/src/main/java/org/springframework/http/client/OutputStreamPublisher.java @@ -30,8 +30,7 @@ import org.springframework.lang.Nullable; import org.springframework.util.Assert; /** - * Bridges between {@link OutputStream} and - * {@link Flow.Publisher Flow.Publisher<T>}. + * Bridges between {@link OutputStream} and {@link Flow.Publisher Flow.Publisher<T>}. * *

Note that this class has a near duplicate in * {@link org.springframework.core.io.buffer.OutputStreamPublisher}. @@ -39,8 +38,7 @@ import org.springframework.util.Assert; * @author Oleh Dokuka * @author Arjen Poutsma * @since 6.1 - * @param the published item type - * @see #create(OutputStreamHandler, ByteMapper, Executor) + * @param the published byte buffer type */ final class OutputStreamPublisher implements Flow.Publisher { @@ -56,98 +54,26 @@ final class OutputStreamPublisher implements Flow.Publisher { private final int chunkSize; - private OutputStreamPublisher( + /** + * Create an instance. + * @param outputStreamHandler invoked when the first buffer is requested + * @param byteMapper maps written bytes to {@code T} + * @param executor used to invoke the {@code outputStreamHandler} + * @param chunkSize the chunk sizes to be produced by the publisher + */ + OutputStreamPublisher( OutputStreamHandler outputStreamHandler, ByteMapper byteMapper, - Executor executor, int chunkSize) { + Executor executor, @Nullable Integer chunkSize) { + + Assert.notNull(outputStreamHandler, "OutputStreamHandler must not be null"); + Assert.notNull(byteMapper, "ByteMapper must not be null"); + Assert.notNull(executor, "Executor must not be null"); + Assert.isTrue(chunkSize == null || chunkSize > 0, "ChunkSize must be larger than 0"); this.outputStreamHandler = outputStreamHandler; this.byteMapper = byteMapper; this.executor = executor; - this.chunkSize = chunkSize; - } - - - /** - * Creates a new {@code Publisher} based on bytes written to a - * {@code OutputStream}. The parameter {@code byteMapper} is used to map - * from written bytes to the published type. - *

    - *
  • The parameter {@code outputStreamHandler} is invoked once per - * subscription of the returned {@code Publisher}, when the first - * item is - * {@linkplain Flow.Subscription#request(long) requested}.
    • - *
  • {@link OutputStream#write(byte[], int, int) OutputStream.write()} - * invocations made by {@code outputStreamHandler} are buffered until they - * exceed the default chunk size of 1024, and then result in a - * {@linkplain Flow.Subscriber#onNext(Object) published} item - * if there is {@linkplain Flow.Subscription#request(long) demand}.
    • - *
  • If there is no demand, {@code OutputStream.write()} will block - * until there is.
    • - *
  • If the subscription is {@linkplain Flow.Subscription#cancel() cancelled}, - * {@code OutputStream.write()} will throw a {@code IOException}.
    • - *
  • The subscription is - * {@linkplain Flow.Subscriber#onComplete() completed} when - * {@code outputStreamHandler} completes.
    • - *
  • Any {@code IOException}s thrown from {@code outputStreamHandler} will - * be dispatched to the {@linkplain Flow.Subscriber#onError(Throwable) Subscriber}. - *
- * @param outputStreamHandler invoked when the first buffer is requested - * @param byteMapper maps written bytes to {@code T} - * @param executor used to invoke the {@code outputStreamHandler} - * @param the publisher type - * @return a {@code Publisher} based on bytes written by - * {@code outputStreamHandler} mapped by {@code byteMapper} - */ - public static Flow.Publisher create(OutputStreamHandler outputStreamHandler, ByteMapper byteMapper, - Executor executor) { - - Assert.notNull(outputStreamHandler, "OutputStreamHandler must not be null"); - Assert.notNull(byteMapper, "ByteMapper must not be null"); - Assert.notNull(executor, "Executor must not be null"); - - return new OutputStreamPublisher<>(outputStreamHandler, byteMapper, executor, DEFAULT_CHUNK_SIZE); - } - - /** - * Creates a new {@code Publisher} based on bytes written to a - * {@code OutputStream}. The parameter {@code byteMapper} is used to map - * from written bytes to the published type. - *
    - *
  • The parameter {@code outputStreamHandler} is invoked once per - * subscription of the returned {@code Publisher}, when the first - * item is - * {@linkplain Flow.Subscription#request(long) requested}.
    • - *
  • {@link OutputStream#write(byte[], int, int) OutputStream.write()} - * invocations made by {@code outputStreamHandler} are buffered until they - * exceed {@code chunkSize}, and then result in a - * {@linkplain Flow.Subscriber#onNext(Object) published} item - * if there is {@linkplain Flow.Subscription#request(long) demand}.
    • - *
  • If there is no demand, {@code OutputStream.write()} will block - * until there is.
    • - *
  • If the subscription is {@linkplain Flow.Subscription#cancel() cancelled}, - * {@code OutputStream.write()} will throw a {@code IOException}.
    • - *
  • The subscription is - * {@linkplain Flow.Subscriber#onComplete() completed} when - * {@code outputStreamHandler} completes.
    • - *
  • Any {@code IOException}s thrown from {@code outputStreamHandler} will - * be dispatched to the {@linkplain Flow.Subscriber#onError(Throwable) Subscriber}. - *
- * @param outputStreamHandler invoked when the first buffer is requested - * @param byteMapper maps written bytes to {@code T} - * @param executor used to invoke the {@code outputStreamHandler} - * @param the publisher type - * @return a {@code Publisher} based on bytes written by - * {@code outputStreamHandler} mapped by {@code byteMapper} - */ - public static Flow.Publisher create(OutputStreamHandler outputStreamHandler, ByteMapper byteMapper, - Executor executor, int chunkSize) { - - Assert.notNull(outputStreamHandler, "OutputStreamHandler must not be null"); - Assert.notNull(byteMapper, "ByteMapper must not be null"); - Assert.notNull(executor, "Executor must not be null"); - Assert.isTrue(chunkSize > 0, "ChunkSize must be larger than 0"); - - return new OutputStreamPublisher<>(outputStreamHandler, byteMapper, executor, chunkSize); + this.chunkSize = (chunkSize != null ? chunkSize : DEFAULT_CHUNK_SIZE); } diff --git a/spring-web/src/main/java/org/springframework/http/client/ReactorClientHttpRequest.java b/spring-web/src/main/java/org/springframework/http/client/ReactorClientHttpRequest.java index 14d1236139..3d38e18aa3 100644 --- a/spring-web/src/main/java/org/springframework/http/client/ReactorClientHttpRequest.java +++ b/spring-web/src/main/java/org/springframework/http/client/ReactorClientHttpRequest.java @@ -113,14 +113,14 @@ final class ReactorClientHttpRequest extends AbstractStreamingClientHttpRequest headers.forEach((key, value) -> reactorRequest.requestHeaders().set(key, value)); if (body != null) { + ByteBufMapper byteMapper = new ByteBufMapper(nettyOutbound.alloc()); AtomicReference executor = new AtomicReference<>(); return nettyOutbound .withConnection(connection -> executor.set(connection.channel().eventLoop())) - .send(FlowAdapters.toPublisher(OutputStreamPublisher.create( - outputStream -> body.writeTo(StreamUtils.nonClosing(outputStream)), - new ByteBufMapper(nettyOutbound.alloc()), - executor.getAndSet(null)))); + .send(FlowAdapters.toPublisher(new OutputStreamPublisher<>( + os -> body.writeTo(StreamUtils.nonClosing(os)), byteMapper, + executor.getAndSet(null), null))); } else { return nettyOutbound; diff --git a/spring-web/src/test/java/org/springframework/http/client/OutputStreamPublisherTests.java b/spring-web/src/test/java/org/springframework/http/client/OutputStreamPublisherTests.java index 466e518e58..b33b1f693f 100644 --- a/spring-web/src/test/java/org/springframework/http/client/OutputStreamPublisherTests.java +++ b/spring-web/src/test/java/org/springframework/http/client/OutputStreamPublisherTests.java @@ -64,11 +64,11 @@ class OutputStreamPublisherTests { @Test void basic() { - Flow.Publisher flowPublisher = OutputStreamPublisher.create(outputStream -> { + Flow.Publisher flowPublisher = new OutputStreamPublisher<>(outputStream -> { outputStream.write(FOO); outputStream.write(BAR); outputStream.write(BAZ); - }, this.byteMapper, this.executor); + }, this.byteMapper, this.executor, null); Flux flux = toString(flowPublisher); StepVerifier.create(flux) @@ -78,14 +78,14 @@ class OutputStreamPublisherTests { @Test void flush() { - Flow.Publisher flowPublisher = OutputStreamPublisher.create(outputStream -> { + Flow.Publisher flowPublisher = new OutputStreamPublisher<>(outputStream -> { outputStream.write(FOO); outputStream.flush(); outputStream.write(BAR); outputStream.flush(); outputStream.write(BAZ); outputStream.flush(); - }, this.byteMapper, this.executor); + }, this.byteMapper, this.executor, null); Flux flux = toString(flowPublisher); StepVerifier.create(flux) @@ -97,7 +97,7 @@ class OutputStreamPublisherTests { @Test void chunkSize() { - Flow.Publisher flowPublisher = OutputStreamPublisher.create(outputStream -> { + Flow.Publisher flowPublisher = new OutputStreamPublisher<>(outputStream -> { outputStream.write(FOO); outputStream.write(BAR); outputStream.write(BAZ); @@ -115,7 +115,7 @@ class OutputStreamPublisherTests { void cancel() throws InterruptedException { CountDownLatch latch = new CountDownLatch(1); - Flow.Publisher flowPublisher = OutputStreamPublisher.create(outputStream -> { + Flow.Publisher flowPublisher = new OutputStreamPublisher<>(outputStream -> { assertThatIOException() .isThrownBy(() -> { outputStream.write(FOO); @@ -126,7 +126,7 @@ class OutputStreamPublisherTests { .withMessage("Subscription has been terminated"); latch.countDown(); - }, this.byteMapper, this.executor); + }, this.byteMapper, this.executor, null); Flux flux = toString(flowPublisher); StepVerifier.create(flux, 1) @@ -141,14 +141,14 @@ class OutputStreamPublisherTests { void closed() throws InterruptedException { CountDownLatch latch = new CountDownLatch(1); - Flow.Publisher flowPublisher = OutputStreamPublisher.create(outputStream -> { + Flow.Publisher flowPublisher = new OutputStreamPublisher<>(outputStream -> { OutputStreamWriter writer = new OutputStreamWriter(outputStream, StandardCharsets.UTF_8); writer.write("foo"); writer.close(); assertThatIOException().isThrownBy(() -> writer.write("bar")) .withMessage("Stream closed"); latch.countDown(); - }, this.byteMapper, this.executor); + }, this.byteMapper, this.executor, null); Flux flux = toString(flowPublisher); StepVerifier.create(flux) @@ -162,7 +162,7 @@ class OutputStreamPublisherTests { void negativeRequestN() throws InterruptedException { CountDownLatch latch = new CountDownLatch(1); - Flow.Publisher flowPublisher = OutputStreamPublisher.create(outputStream -> { + Flow.Publisher flowPublisher = new OutputStreamPublisher<>(outputStream -> { try (outputStream) { outputStream.write(FOO); outputStream.flush(); @@ -172,7 +172,7 @@ class OutputStreamPublisherTests { finally { latch.countDown(); } - }, this.byteMapper, this.executor); + }, this.byteMapper, this.executor, null); Flow.Subscription[] subscriptions = new Flow.Subscription[1]; Flux flux = toString(a-> flowPublisher.subscribe(new Flow.Subscriber<>() { @Override