Rename "ResponseBody" flush processor + use generics

Issue: SPR-14527

spring-web/src/main/java/org/springframework/http/server/reactive/AbstractListenerFlushProcessor.java

@@ -27,8 +27,6 @@ import org.reactivestreams.Publisher;
 import org.reactivestreams.Subscriber;
 import org.reactivestreams.Subscription;
 
-import org.springframework.core.io.buffer.DataBuffer;
-
 /**
  * Abstract base class for {@code Processor} implementations that bridge between
  * event-listener APIs and Reactive Streams. Specifically, base class for the
@@ -41,11 +39,11 @@ import org.springframework.core.io.buffer.DataBuffer;
  * @see UndertowHttpHandlerAdapter
  * @see ServerHttpResponse#writeAndFlushWith(Publisher)
  */
-abstract class AbstractResponseBodyFlushProcessor implements Processor<Publisher<? extends DataBuffer>, Void> {
+public abstract class AbstractListenerFlushProcessor<T> implements Processor<Publisher<? extends T>, Void> {
 
 	protected final Log logger = LogFactory.getLog(getClass());
 
-	private final ResponseBodyWriteResultPublisher resultPublisher = new ResponseBodyWriteResultPublisher();
+	private final WriteResultPublisher resultPublisher = new WriteResultPublisher();
 
 	private final AtomicReference<State> state = new AtomicReference<>(State.UNSUBSCRIBED);
 
@@ -65,7 +63,7 @@ abstract class AbstractResponseBodyFlushProcessor implements Processor<Publisher
 	}
 
 	@Override
-	public final void onNext(Publisher<? extends DataBuffer> publisher) {
+	public final void onNext(Publisher<? extends T> publisher) {
 		if (logger.isTraceEnabled()) {
 			logger.trace(this.state + " onNext: " + publisher);
 		}
@@ -100,7 +98,7 @@ abstract class AbstractResponseBodyFlushProcessor implements Processor<Publisher
 	/**
 	 * Creates a new processor for subscribing to a body chunk.
 	 */
-	protected abstract Processor<? super DataBuffer, Void> createBodyProcessor();
+	protected abstract Processor<? super T, Void> createBodyProcessor();
 
 	/**
 	 * Flushes the output.
@@ -130,7 +128,7 @@ abstract class AbstractResponseBodyFlushProcessor implements Processor<Publisher
 		UNSUBSCRIBED {
 
 			@Override
-			public void onSubscribe(AbstractResponseBodyFlushProcessor processor, Subscription subscription) {
+			public <T> void onSubscribe(AbstractListenerFlushProcessor<T> processor, Subscription subscription) {
 				Objects.requireNonNull(subscription, "Subscription cannot be null");
 				if (processor.changeState(this, REQUESTED)) {
 					processor.subscription = subscription;
@@ -144,16 +142,16 @@ abstract class AbstractResponseBodyFlushProcessor implements Processor<Publisher
 		REQUESTED {
 
 			@Override
-			public void onNext(AbstractResponseBodyFlushProcessor processor, Publisher<? extends DataBuffer> chunk) {
+			public <T> void onNext(AbstractListenerFlushProcessor<T> processor, Publisher<? extends T> chunk) {
 				if (processor.changeState(this, RECEIVED)) {
-					Processor<? super DataBuffer, Void> chunkProcessor = processor.createBodyProcessor();
+					Processor<? super T, Void> chunkProcessor = processor.createBodyProcessor();
 					chunk.subscribe(chunkProcessor);
 					chunkProcessor.subscribe(new WriteSubscriber(processor));
 				}
 			}
 
 			@Override
-			public void onComplete(AbstractResponseBodyFlushProcessor processor) {
+			public <T> void onComplete(AbstractListenerFlushProcessor<T> processor) {
 				if (processor.changeState(this, COMPLETED)) {
 					processor.resultPublisher.publishComplete();
 				}
@@ -162,7 +160,7 @@ abstract class AbstractResponseBodyFlushProcessor implements Processor<Publisher
 		RECEIVED {
 
 			@Override
-			public void writeComplete(AbstractResponseBodyFlushProcessor processor) {
+			public <T> void writeComplete(AbstractListenerFlushProcessor<T> processor) {
 				try {
 					processor.flush();
 				}
@@ -184,58 +182,58 @@ abstract class AbstractResponseBodyFlushProcessor implements Processor<Publisher
 			}
 
 			@Override
-			public void onComplete(AbstractResponseBodyFlushProcessor processor) {
+			public <T> void onComplete(AbstractListenerFlushProcessor<T> processor) {
 				processor.subscriberCompleted = true;
 			}
 		},
 		COMPLETED {
 
 			@Override
-			public void onNext(AbstractResponseBodyFlushProcessor processor,
-					Publisher<? extends DataBuffer> publisher) {
+			public <T> void onNext(AbstractListenerFlushProcessor<T> processor,
+					Publisher<? extends T> publisher) {
 				// ignore
 
 			}
 
 			@Override
-			public void onError(AbstractResponseBodyFlushProcessor processor, Throwable t) {
+			public <T> void onError(AbstractListenerFlushProcessor<T> processor, Throwable t) {
 				// ignore
 			}
 
 			@Override
-			public void onComplete(AbstractResponseBodyFlushProcessor processor) {
+			public <T> void onComplete(AbstractListenerFlushProcessor<T> processor) {
 				// ignore
 			}
 		};
 
-		public void onSubscribe(AbstractResponseBodyFlushProcessor processor, Subscription subscription) {
+		public <T> void onSubscribe(AbstractListenerFlushProcessor<T> processor, Subscription subscription) {
 			subscription.cancel();
 		}
 
-		public void onNext(AbstractResponseBodyFlushProcessor processor, Publisher<? extends DataBuffer> publisher) {
+		public <T> void onNext(AbstractListenerFlushProcessor<T> processor, Publisher<? extends T> publisher) {
 			throw new IllegalStateException(toString());
 		}
 
-		public void onError(AbstractResponseBodyFlushProcessor processor, Throwable ex) {
+		public <T> void onError(AbstractListenerFlushProcessor<T> processor, Throwable ex) {
 			if (processor.changeState(this, COMPLETED)) {
 				processor.resultPublisher.publishError(ex);
 			}
 		}
 
-		public void onComplete(AbstractResponseBodyFlushProcessor processor) {
+		public <T> void onComplete(AbstractListenerFlushProcessor<T> processor) {
 			throw new IllegalStateException(toString());
 		}
 
-		public void writeComplete(AbstractResponseBodyFlushProcessor processor) {
+		public <T> void writeComplete(AbstractListenerFlushProcessor<T> processor) {
 			// ignore
 		}
 
 
 		private static class WriteSubscriber implements Subscriber<Void> {
 
-			private final AbstractResponseBodyFlushProcessor processor;
+			private final AbstractListenerFlushProcessor<?> processor;
 
-			public WriteSubscriber(AbstractResponseBodyFlushProcessor processor) {
+			public WriteSubscriber(AbstractListenerFlushProcessor<?> processor) {
 				this.processor = processor;
 			}
 

spring-web/src/main/java/org/springframework/http/server/reactive/AbstractListenerWriteProcessor.java

@@ -48,7 +48,7 @@ public abstract class AbstractListenerWriteProcessor<T> implements Processor<T,
 
 	protected final Log logger = LogFactory.getLog(getClass());
 
-	private final ResponseBodyWriteResultPublisher resultPublisher = new ResponseBodyWriteResultPublisher();
+	private final WriteResultPublisher resultPublisher = new WriteResultPublisher();
 
 	private final AtomicReference<State> state = new AtomicReference<>(State.UNSUBSCRIBED);
 

spring-web/src/main/java/org/springframework/http/server/reactive/ServletServerHttpResponse.java

@@ -273,7 +273,7 @@ public class ServletServerHttpResponse extends AbstractListenerServerHttpRespons
 	}
 
 
-	private class ResponseBodyFlushProcessor extends AbstractResponseBodyFlushProcessor {
+	private class ResponseBodyFlushProcessor extends AbstractListenerFlushProcessor<DataBuffer> {
 
 		@Override
 		protected Processor<? super DataBuffer, Void> createBodyProcessor() {

spring-web/src/main/java/org/springframework/http/server/reactive/UndertowServerHttpResponse.java

@@ -218,7 +218,7 @@ public class UndertowServerHttpResponse extends AbstractListenerServerHttpRespon
 		}
 	}
 
-	private class ResponseBodyFlushProcessor extends AbstractResponseBodyFlushProcessor {
+	private class ResponseBodyFlushProcessor extends AbstractListenerFlushProcessor<DataBuffer> {
 
 		@Override
 		protected Processor<? super DataBuffer, Void> createBodyProcessor() {

spring-web/src/main/java/org/springframework/http/server/reactive/WriteResultPublisher.java

@@ -30,11 +30,12 @@ import reactor.core.publisher.Operators;
  * Publisher returned from {@link ServerHttpResponse#writeWith(Publisher)}.
  *
  * @author Arjen Poutsma
+ * @author Violeta Georgieva
  * @since 5.0
  */
-class ResponseBodyWriteResultPublisher implements Publisher<Void> {
+class WriteResultPublisher implements Publisher<Void> {
 
-	private static final Log logger = LogFactory.getLog(ResponseBodyWriteResultPublisher.class);
+	private static final Log logger = LogFactory.getLog(WriteResultPublisher.class);
 
 	private final AtomicReference<State> state = new AtomicReference<>(State.UNSUBSCRIBED);
 
@@ -80,9 +81,9 @@ class ResponseBodyWriteResultPublisher implements Publisher<Void> {
 
 	private static final class ResponseBodyWriteResultSubscription implements Subscription {
 
-		private final ResponseBodyWriteResultPublisher publisher;
+		private final WriteResultPublisher publisher;
 
-		public ResponseBodyWriteResultSubscription(ResponseBodyWriteResultPublisher publisher) {
+		public ResponseBodyWriteResultSubscription(WriteResultPublisher publisher) {
 			this.publisher = publisher;
 		}
 
@@ -112,7 +113,7 @@ class ResponseBodyWriteResultPublisher implements Publisher<Void> {
 
 		UNSUBSCRIBED {
 			@Override
-			void subscribe(ResponseBodyWriteResultPublisher publisher,
+			void subscribe(WriteResultPublisher publisher,
 					Subscriber<? super Void> subscriber) {
 				Objects.requireNonNull(subscriber);
 				if (publisher.changeState(this, SUBSCRIBED)) {
@@ -132,28 +133,28 @@ class ResponseBodyWriteResultPublisher implements Publisher<Void> {
 				}
 			}
 			@Override
-			void publishComplete(ResponseBodyWriteResultPublisher publisher) {
+			void publishComplete(WriteResultPublisher publisher) {
 				publisher.publisherCompleted = true;
 			}
 			@Override
-			void publishError(ResponseBodyWriteResultPublisher publisher, Throwable t) {
+			void publishError(WriteResultPublisher publisher, Throwable t) {
 				publisher.publisherError = t;
 			}
 		},
 
 		SUBSCRIBED {
 			@Override
-			void request(ResponseBodyWriteResultPublisher publisher, long n) {
+			void request(WriteResultPublisher publisher, long n) {
 				Operators.checkRequest(n, publisher.subscriber);
 			}
 			@Override
-			void publishComplete(ResponseBodyWriteResultPublisher publisher) {
+			void publishComplete(WriteResultPublisher publisher) {
 				if (publisher.changeState(this, COMPLETED)) {
 					publisher.subscriber.onComplete();
 				}
 			}
 			@Override
-			void publishError(ResponseBodyWriteResultPublisher publisher, Throwable t) {
+			void publishError(WriteResultPublisher publisher, Throwable t) {
 				if (publisher.changeState(this, COMPLETED)) {
 					publisher.subscriber.onError(t);
 				}
@@ -162,40 +163,40 @@ class ResponseBodyWriteResultPublisher implements Publisher<Void> {
 
 		COMPLETED {
 			@Override
-			void request(ResponseBodyWriteResultPublisher publisher, long n) {
+			void request(WriteResultPublisher publisher, long n) {
 				// ignore
 			}
 			@Override
-			void cancel(ResponseBodyWriteResultPublisher publisher) {
+			void cancel(WriteResultPublisher publisher) {
 				// ignore
 			}
 			@Override
-			void publishComplete(ResponseBodyWriteResultPublisher publisher) {
+			void publishComplete(WriteResultPublisher publisher) {
 				// ignore
 			}
 			@Override
-			void publishError(ResponseBodyWriteResultPublisher publisher, Throwable t) {
+			void publishError(WriteResultPublisher publisher, Throwable t) {
 				// ignore
 			}
 		};
 
-		void subscribe(ResponseBodyWriteResultPublisher publisher, Subscriber<? super Void> subscriber) {
+		void subscribe(WriteResultPublisher publisher, Subscriber<? super Void> subscriber) {
 			throw new IllegalStateException(toString());
 		}
 
-		void request(ResponseBodyWriteResultPublisher publisher, long n) {
+		void request(WriteResultPublisher publisher, long n) {
 			throw new IllegalStateException(toString());
 		}
 
-		void cancel(ResponseBodyWriteResultPublisher publisher) {
+		void cancel(WriteResultPublisher publisher) {
 			publisher.changeState(this, COMPLETED);
 		}
 
-		void publishComplete(ResponseBodyWriteResultPublisher publisher) {
+		void publishComplete(WriteResultPublisher publisher) {
 			throw new IllegalStateException(toString());
 		}
 
-		void publishError(ResponseBodyWriteResultPublisher publisher, Throwable t) {
+		void publishError(WriteResultPublisher publisher, Throwable t) {
 			throw new IllegalStateException(toString());
 		}
 	}