Align OutputStreamPublisher's

Apply recent nullability changes, and formatting differences.

See gh-33592

Author: rstoyanchev

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

@@ -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 @@ private static final class OutputStreamSubscription extends OutputStream impleme
 
 		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;

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

@@ -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 @@ public void subscribe(Flow.Subscriber<? super T> subscriber) {
 		// 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 @@ private static final class OutputStreamSubscription<T> extends OutputStream impl
 
 		static final Object READY = new Object();
 
-
 		private final Flow.Subscriber<? super T> actual;
 
 		private final OutputStreamHandler outputStreamHandler;
@@ -236,12 +239,13 @@ private static final class OutputStreamSubscription<T> extends OutputStream impl
 
 		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 @@ private void invokeHandler() {
 				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 @@ private void invokeHandler() {
 			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 @@ public void request(long n) {
 			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 @@ private void resume() {
 		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 @@ private long tryCancel() {
 		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 @@ private static long addCap(long a, long b) {
 			return res;
 		}
 	}
+
 }