From 113b430fab068d960abe698acdffc65d8901fda9 Mon Sep 17 00:00:00 2001 From: rstoyanchev Date: Tue, 24 Sep 2024 15:41:37 +0100 Subject: [PATCH] Align OutputStreamPublisher's Apply recent nullability changes, and formatting differences. See gh-33592 --- .../core/io/buffer/OutputStreamPublisher.java | 8 ++-- .../http/client/OutputStreamPublisher.java | 44 +++++++++---------- 2 files changed, 27 insertions(+), 25 deletions(-) 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 6b5eb67abf..8e77f26eaa 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 @@ -54,7 +54,8 @@ final class OutputStreamPublisher implements Publisher { private final int chunkSize; - OutputStreamPublisher(Consumer outputStreamConsumer, DataBufferFactory bufferFactory, + OutputStreamPublisher( + Consumer outputStreamConsumer, DataBufferFactory bufferFactory, Executor executor, int chunkSize) { this.outputStreamConsumer = outputStreamConsumer; @@ -99,8 +100,9 @@ final class OutputStreamPublisher implements Publisher { private long produced; - OutputStreamSubscription(Subscriber actual, - Consumer outputStreamConsumer, DataBufferFactory bufferFactory, int chunkSize) { + OutputStreamSubscription( + Subscriber actual, Consumer outputStreamConsumer, + DataBufferFactory bufferFactory, int chunkSize) { this.actual = actual; this.outputStreamHandler = outputStreamConsumer; 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 84eb00e8d9..6bfde107c9 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 @@ -56,7 +56,10 @@ final class OutputStreamPublisher implements Flow.Publisher { private final int chunkSize; - private OutputStreamPublisher(OutputStreamHandler outputStreamHandler, ByteMapper byteMapper, Executor executor, int chunkSize) { + private OutputStreamPublisher( + OutputStreamHandler outputStreamHandler, ByteMapper byteMapper, + Executor executor, int chunkSize) { + this.outputStreamHandler = outputStreamHandler; this.byteMapper = byteMapper; this.executor = executor; @@ -154,8 +157,9 @@ final class OutputStreamPublisher implements Flow.Publisher { // for Reactive Streams compliance. Objects.requireNonNull(subscriber, "Subscriber must not be null"); - OutputStreamSubscription subscription = new OutputStreamSubscription<>(subscriber, this.outputStreamHandler, - this.byteMapper, this.chunkSize); + OutputStreamSubscription 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 implements Flow.Publisher { static final Object READY = new Object(); - private final Flow.Subscriber actual; private final OutputStreamHandler outputStreamHandler; @@ -236,12 +239,13 @@ final class OutputStreamPublisher implements Flow.Publisher { private long produced; - - public OutputStreamSubscription(Flow.Subscriber actual, OutputStreamHandler outputStreamHandler, + OutputStreamSubscription( + Flow.Subscriber actual, OutputStreamHandler outputStreamHandler, ByteMapper 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 implements Flow.Publisher { 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 implements Flow.Publisher { 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 implements Flow.Publisher { 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 implements Flow.Publisher { 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 implements Flow.Publisher { 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 implements Flow.Publisher { return res; } } + }