Spring/spring-framework
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Align OutputStreamPublisher's

Browse Source 
Update constructors to match and drop unnecessary factory methods.

See gh-33592
This commit is contained in:
rstoyanchev
2024-09-24 16:01:22 +01:00
parent 113b430fab
commit f6c31bb6c3
6 changed files with 58 additions and 151 deletions
Download Patch File Download Diff File Expand all files Collapse all files

39
spring-core/src/main/java/org/springframework/core/io/buffer/DataBufferUtils.java
View File

@@ -68,8 +68,6 @@ public abstract class DataBufferUtils {
private static final Consumer<DataBuffer> RELEASE_CONSUMER = DataBufferUtils::release;
private static final int DEFAULT_CHUNK_SIZE = 1024;
//---------------------------------------------------------------------
// Reading
@@ -442,48 +440,17 @@ public abstract class DataBufferUtils {
public static Publisher<DataBuffer> outputStreamPublisher(Consumer<OutputStream> 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<DataBuffer>} based on bytes written to a
* {@code OutputStream}.
* <ul>
* <li>The parameter {@code outputStreamConsumer} is invoked once per
* subscription of the returned {@code Publisher}, when the first
* item is
* {@linkplain Subscription#request(long) requested}.</li>
* <li>{@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}.</li>
* <li>If there is <em>no demand</em>, {@code OutputStream.write()} will block
* until there is.</li>
* <li>If the subscription is {@linkplain Subscription#cancel() cancelled},
* {@code OutputStream.write()} will throw a {@code IOException}.</li>
* <li>The subscription is
* {@linkplain Subscriber#onComplete() completed} when
* {@code outputStreamHandler} completes.</li>
* <li>Any exceptions thrown from {@code outputStreamHandler} will
* be dispatched to the {@linkplain Subscriber#onError(Throwable) Subscriber}.
* </ul>
* @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<DataBuffer>} 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<DataBuffer> outputStreamPublisher(Consumer<OutputStream> 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);
}

23
spring-core/src/main/java/org/springframework/core/io/buffer/OutputStreamPublisher.java
View File

@@ -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&lt;DataBuffer&gt;}.
* Bridges between {@link OutputStream} and {@link Publisher Publisher&lt;DataBuffer&gt;}.
*
* <p>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<DataBuffer> {
private static final int DEFAULT_CHUNK_SIZE = 1024;
private final Consumer<OutputStream> outputStreamConsumer;
private final DataBufferFactory bufferFactory;
@@ -54,14 +57,26 @@ final class OutputStreamPublisher implements Publisher<DataBuffer> {
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<OutputStream> 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);
}

9
spring-web/src/main/java/org/springframework/http/client/JdkClientHttpRequest.java
View File

@@ -153,19 +153,18 @@ class JdkClientHttpRequest extends AbstractStreamingClientHttpRequest {
private HttpRequest.BodyPublisher bodyPublisher(HttpHeaders headers, @Nullable Body body) {
if (body != null) {
Flow.Publisher<ByteBuffer> outputStreamPublisher = OutputStreamPublisher.create(
outputStream -> body.writeTo(StreamUtils.nonClosing(outputStream)),
BYTE_MAPPER, this.executor);
Flow.Publisher<ByteBuffer> 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 {

108
spring-web/src/main/java/org/springframework/http/client/OutputStreamPublisher.java
View File

@@ -30,8 +30,7 @@ import org.springframework.lang.Nullable;
import org.springframework.util.Assert;
/**
* Bridges between {@link OutputStream} and
* {@link Flow.Publisher Flow.Publisher&lt;T&gt;}.
* Bridges between {@link OutputStream} and {@link Flow.Publisher Flow.Publisher&lt;T&gt;}.
*
* <p>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 <T> the published item type
* @see #create(OutputStreamHandler, ByteMapper, Executor)
* @param <T> the published byte buffer type
*/
final class OutputStreamPublisher<T> implements Flow.Publisher<T> {
@@ -56,98 +54,26 @@ final class OutputStreamPublisher<T> implements Flow.Publisher<T> {
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<T> 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<T>} based on bytes written to a
* {@code OutputStream}. The parameter {@code byteMapper} is used to map
* from written bytes to the published type.
* <ul>
* <li>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}.</li>
* <li>{@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}.</li>
* <li>If there is <em>no demand</em>, {@code OutputStream.write()} will block
* until there is.</li>
* <li>If the subscription is {@linkplain Flow.Subscription#cancel() cancelled},
* {@code OutputStream.write()} will throw a {@code IOException}.</li>
* <li>The subscription is
* {@linkplain Flow.Subscriber#onComplete() completed} when
* {@code outputStreamHandler} completes.</li>
* <li>Any {@code IOException}s thrown from {@code outputStreamHandler} will
* be dispatched to the {@linkplain Flow.Subscriber#onError(Throwable) Subscriber}.
* </ul>
* @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 <T> the publisher type
* @return a {@code Publisher<T>} based on bytes written by
* {@code outputStreamHandler} mapped by {@code byteMapper}
*/
public static <T> Flow.Publisher<T> create(OutputStreamHandler outputStreamHandler, ByteMapper<T> 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<T>} based on bytes written to a
* {@code OutputStream}. The parameter {@code byteMapper} is used to map
* from written bytes to the published type.
* <ul>
* <li>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}.</li>
* <li>{@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}.</li>
* <li>If there is <em>no demand</em>, {@code OutputStream.write()} will block
* until there is.</li>
* <li>If the subscription is {@linkplain Flow.Subscription#cancel() cancelled},
* {@code OutputStream.write()} will throw a {@code IOException}.</li>
* <li>The subscription is
* {@linkplain Flow.Subscriber#onComplete() completed} when
* {@code outputStreamHandler} completes.</li>
* <li>Any {@code IOException}s thrown from {@code outputStreamHandler} will
* be dispatched to the {@linkplain Flow.Subscriber#onError(Throwable) Subscriber}.
* </ul>
* @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 <T> the publisher type
* @return a {@code Publisher<T>} based on bytes written by
* {@code outputStreamHandler} mapped by {@code byteMapper}
*/
public static <T> Flow.Publisher<T> create(OutputStreamHandler outputStreamHandler, ByteMapper<T> 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);
}

8
spring-web/src/main/java/org/springframework/http/client/ReactorClientHttpRequest.java
View File

@@ -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> 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;

22
spring-web/src/test/java/org/springframework/http/client/OutputStreamPublisherTests.java
View File

@@ -64,11 +64,11 @@ class OutputStreamPublisherTests {
@Test
void basic() {
Flow.Publisher<byte[]> flowPublisher = OutputStreamPublisher.create(outputStream -> {
Flow.Publisher<byte[]> flowPublisher = new OutputStreamPublisher<>(outputStream -> {
outputStream.write(FOO);
outputStream.write(BAR);
outputStream.write(BAZ);
}, this.byteMapper, this.executor);
}, this.byteMapper, this.executor, null);
Flux<String> flux = toString(flowPublisher);
StepVerifier.create(flux)
@@ -78,14 +78,14 @@ class OutputStreamPublisherTests {
@Test
void flush() {
Flow.Publisher<byte[]> flowPublisher = OutputStreamPublisher.create(outputStream -> {
Flow.Publisher<byte[]> 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<String> flux = toString(flowPublisher);
StepVerifier.create(flux)
@@ -97,7 +97,7 @@ class OutputStreamPublisherTests {
@Test
void chunkSize() {
Flow.Publisher<byte[]> flowPublisher = OutputStreamPublisher.create(outputStream -> {
Flow.Publisher<byte[]> 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<byte[]> flowPublisher = OutputStreamPublisher.create(outputStream -> {
Flow.Publisher<byte[]> 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<String> flux = toString(flowPublisher);
StepVerifier.create(flux, 1)
@@ -141,14 +141,14 @@ class OutputStreamPublisherTests {
void closed() throws InterruptedException {
CountDownLatch latch = new CountDownLatch(1);
Flow.Publisher<byte[]> flowPublisher = OutputStreamPublisher.create(outputStream -> {
Flow.Publisher<byte[]> 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<String> flux = toString(flowPublisher);
StepVerifier.create(flux)
@@ -162,7 +162,7 @@ class OutputStreamPublisherTests {
void negativeRequestN() throws InterruptedException {
CountDownLatch latch = new CountDownLatch(1);
Flow.Publisher<byte[]> flowPublisher = OutputStreamPublisher.create(outputStream -> {
Flow.Publisher<byte[]> 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<String> flux = toString(a-> flowPublisher.subscribe(new Flow.Subscriber<>() {
@Override