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

Align OutputStreamPublisher's

Browse Source 
Apply recent nullability changes, and formatting differences.

See gh-33592
This commit is contained in:
rstoyanchev
2024-09-24 15:41:37 +01:00
parent a1613d1352
commit 113b430fab
2 changed files with 27 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -54,7 +54,8 @@ final class OutputStreamPublisher implements Publisher<DataBuffer> {
private final int chunkSize;
OutputStreamPublisher(Consumer<OutputStream> outputStreamConsumer, DataBufferFactory bufferFactory,
OutputStreamPublisher(
Consumer<OutputStream> outputStreamConsumer, DataBufferFactory bufferFactory,
Executor executor, int chunkSize) {
this.outputStreamConsumer = outputStreamConsumer;
@@ -99,8 +100,9 @@ final class OutputStreamPublisher implements Publisher<DataBuffer> {
private long produced;
OutputStreamSubscription(Subscriber<? super DataBuffer> actual,
Consumer<OutputStream> outputStreamConsumer, DataBufferFactory bufferFactory, int chunkSize) {
OutputStreamSubscription(
Subscriber<? super DataBuffer> actual, Consumer<OutputStream> outputStreamConsumer,
DataBufferFactory bufferFactory, int chunkSize) {
this.actual = actual;
this.outputStreamHandler = outputStreamConsumer;

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

@@ -56,7 +56,10 @@ final class OutputStreamPublisher<T> implements Flow.Publisher<T> {
private final int chunkSize;
private OutputStreamPublisher(OutputStreamHandler outputStreamHandler, ByteMapper<T> byteMapper, Executor executor, int chunkSize) {
private OutputStreamPublisher(
OutputStreamHandler outputStreamHandler, ByteMapper<T> byteMapper,
Executor executor, int chunkSize) {
this.outputStreamHandler = outputStreamHandler;
this.byteMapper = byteMapper;
this.executor = executor;
@@ -154,8 +157,9 @@ final class OutputStreamPublisher<T> implements Flow.Publisher<T> {
// for Reactive Streams compliance.
Objects.requireNonNull(subscriber, "Subscriber must not be null");
OutputStreamSubscription<T> subscription = new OutputStreamSubscription<>(subscriber, this.outputStreamHandler,
this.byteMapper, this.chunkSize);
OutputStreamSubscription<T> subscription = new OutputStreamSubscription<>(
subscriber, this.outputStreamHandler, this.byteMapper, this.chunkSize);
subscriber.onSubscribe(subscription);
this.executor.execute(subscription::invokeHandler);
}
@@ -218,7 +222,6 @@ final class OutputStreamPublisher<T> implements Flow.Publisher<T> {
static final Object READY = new Object();
private final Flow.Subscriber<? super T> actual;
private final OutputStreamHandler outputStreamHandler;
@@ -236,12 +239,13 @@ final class OutputStreamPublisher<T> implements Flow.Publisher<T> {
private long produced;
public OutputStreamSubscription(Flow.Subscriber<? super T> actual, OutputStreamHandler outputStreamHandler,
OutputStreamSubscription(
Flow.Subscriber<? super T> actual, OutputStreamHandler outputStreamHandler,
ByteMapper<T> byteMapper, int chunkSize) {
this.actual = actual;
this.byteMapper = byteMapper;
this.outputStreamHandler = outputStreamHandler;
this.byteMapper = byteMapper;
this.chunkSize = chunkSize;
}
@@ -315,13 +319,14 @@ final class OutputStreamPublisher<T> implements Flow.Publisher<T> {
if (isCancelled(previousState)) {
return;
}
if (isTerminated(previousState)) {
// failure due to illegal requestN
this.actual.onError(this.error);
return;
Throwable error = this.error;
if (error != null) {
this.actual.onError(error);
return;
}
}
this.actual.onError(ex);
return;
}
@@ -330,13 +335,14 @@ final class OutputStreamPublisher<T> implements Flow.Publisher<T> {
if (isCancelled(previousState)) {
return;
}
if (isTerminated(previousState)) {
// failure due to illegal requestN
this.actual.onError(this.error);
return;
Throwable error = this.error;
if (error != null) {
this.actual.onError(error);
return;
}
}
this.actual.onComplete();
}
@@ -346,16 +352,13 @@ final class OutputStreamPublisher<T> implements Flow.Publisher<T> {
if (n <= 0) {
this.error = new IllegalArgumentException("request should be a positive number");
long previousState = tryTerminate();
if (isTerminated(previousState) || isCancelled(previousState)) {
return;
}
if (previousState > 0) {
// error should eventually be observed and propagated
return;
}
// resume parked thread, so it can observe error and propagate it
resume();
return;
@@ -413,11 +416,9 @@ final class OutputStreamPublisher<T> implements Flow.Publisher<T> {
private long tryCancel() {
while (true) {
long r = this.requested.get();
if (isCancelled(r)) {
return r;
}
if (this.requested.compareAndSet(r, Long.MIN_VALUE)) {
return r;
}
@@ -427,11 +428,9 @@ final class OutputStreamPublisher<T> implements Flow.Publisher<T> {
private long tryTerminate() {
while (true) {
long r = this.requested.get();
if (isCancelled(r) || isTerminated(r)) {
return r;
}
if (this.requested.compareAndSet(r, Long.MIN_VALUE | Long.MAX_VALUE)) {
return r;
}
@@ -486,4 +485,5 @@ final class OutputStreamPublisher<T> implements Flow.Publisher<T> {
return res;
}
}
}