Upgrade to Reactor Core 3.1

Issue: SPR-15318

spring-web/src/main/java/org/springframework/http/codec/FormHttpMessageWriter.java

@@ -97,7 +97,7 @@ public class FormHttpMessageWriter implements HttpMessageWriter<MultiValueMap<St
 				.from(inputStream)
 				.single()
 				.map(form -> generateForm(form, charset))
-				.then(value -> {
+				.flatMap(value -> {
 					ByteBuffer byteBuffer = charset.encode(value);
 					DataBuffer buffer = message.bufferFactory().wrap(byteBuffer);
 					message.getHeaders().setContentLength(byteBuffer.remaining());

spring-web/src/main/java/org/springframework/http/codec/ResourceHttpMessageWriter.java

@@ -108,7 +108,7 @@ public class ResourceHttpMessageWriter implements HttpMessageWriter<Resource> {
 	public Mono<Void> write(Publisher<? extends Resource> inputStream, ResolvableType elementType,
 			MediaType mediaType, ReactiveHttpOutputMessage message, Map<String, Object> hints) {
 
-		return Mono.from(inputStream).then(resource ->
+		return Mono.from(inputStream).flatMap(resource ->
 				writeResource(resource, elementType, mediaType, message, hints));
 	}
 
@@ -191,7 +191,7 @@ public class ResourceHttpMessageWriter implements HttpMessageWriter<Resource> {
 			return response.setComplete();
 		}
 
-		return Mono.from(inputStream).then(resource -> {
+		return Mono.from(inputStream).flatMap(resource -> {
 
 			if (ranges.isEmpty()) {
 				return writeResource(resource, elementType, mediaType, response, hints);

spring-web/src/main/java/org/springframework/http/codec/xml/XmlEventDecoder.java

@@ -99,7 +99,7 @@ public class XmlEventDecoder extends AbstractDecoder<XMLEvent> {
 		else {
 			Mono<DataBuffer> singleBuffer = flux.reduce(DataBuffer::write);
 			return singleBuffer.
-					flatMap(dataBuffer -> {
+					flatMapMany(dataBuffer -> {
 						try {
 							InputStream is = dataBuffer.asInputStream();
 							XMLEventReader eventReader = inputFactory.createXMLEventReader(is);

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

@@ -21,6 +21,7 @@ import java.util.function.Function;
 import org.reactivestreams.Publisher;
 import org.reactivestreams.Subscriber;
 import org.reactivestreams.Subscription;
+import reactor.core.Exceptions;
 import reactor.core.publisher.MonoSource;
 import reactor.core.publisher.Operators;
 
@@ -56,7 +57,7 @@ public class ChannelSendOperator<T> extends MonoSource<T, Void> {
 
 
 	@SuppressWarnings("deprecation")
-	private class WriteWithBarrier extends Operators.SubscriberAdapter<T, Void> implements Publisher<T> {
+	private class WriteWithBarrier extends SubscriberAdapter<T, Void> implements Publisher<T> {
 
 		/**
 		 * We've at at least one emission, we've called the write function, the write
@@ -210,6 +211,134 @@ public class ChannelSendOperator<T> extends MonoSource<T, Void> {
 		}
 	}
 
+	// TODO Remove this copy of Reactor 3.0.x Operators.SubscriberAdapter
+	private static class SubscriberAdapter<I, O> implements Subscriber<I>, Subscription {
+
+		protected final Subscriber<? super O> subscriber;
+
+		protected Subscription subscription;
+
+		public SubscriberAdapter(Subscriber<? super O> subscriber) {
+			this.subscriber = subscriber;
+		}
+
+		public Subscriber<? super O> downstream() {
+			return subscriber;
+		}
+
+		@Override
+		public final void cancel() {
+			try {
+				doCancel();
+			} catch (Throwable throwable) {
+				doOnSubscriberError(Operators.onOperatorError(subscription, throwable));
+			}
+		}
+
+		@Override
+		public final void onComplete() {
+			try {
+				doComplete();
+			} catch (Throwable throwable) {
+				doOnSubscriberError(Operators.onOperatorError(throwable));
+			}
+		}
+
+		@Override
+		public final void onError(Throwable t) {
+			if (t == null) {
+				throw Exceptions.argumentIsNullException();
+			}
+			doError(t);
+		}
+
+		@Override
+		public final void onNext(I i) {
+			if (i == null) {
+				throw Exceptions.argumentIsNullException();
+			}
+			try {
+				doNext(i);
+			}
+			catch (Throwable throwable) {
+				doOnSubscriberError(Operators.onOperatorError(subscription, throwable, i));
+			}
+		}
+
+		@Override
+		public final void onSubscribe(Subscription s) {
+			if (Operators.validate(subscription, s)) {
+				try {
+					subscription = s;
+					doOnSubscribe(s);
+				}
+				catch (Throwable throwable) {
+					doOnSubscriberError(Operators.onOperatorError(s, throwable));
+				}
+			}
+		}
+
+		@Override
+		public final void request(long n) {
+			try {
+				Operators.checkRequest(n);
+				doRequest(n);
+			} catch (Throwable throwable) {
+				doCancel();
+				doOnSubscriberError(Operators.onOperatorError(throwable));
+			}
+		}
+
+		@Override
+		public String toString() {
+			return getClass().getSimpleName();
+		}
+
+		/**
+		 * Hook for further processing of onSubscribe's Subscription.
+		 * @param subscription the subscription to optionally process
+		 */
+		protected void doOnSubscribe(Subscription subscription) {
+			subscriber.onSubscribe(this);
+		}
+
+		public Subscription upstream() {
+			return subscription;
+		}
+
+		@SuppressWarnings("unchecked")
+		protected void doNext(I i) {
+			subscriber.onNext((O) i);
+		}
+
+		protected void doError(Throwable throwable) {
+			subscriber.onError(throwable);
+		}
+
+		protected void doOnSubscriberError(Throwable throwable){
+			subscriber.onError(throwable);
+		}
+
+		protected void doComplete() {
+			subscriber.onComplete();
+		}
+
+		protected void doRequest(long n) {
+			Subscription s = this.subscription;
+			if (s != null) {
+				s.request(n);
+			}
+		}
+
+		protected void doCancel() {
+			Subscription s = this.subscription;
+			if (s != null) {
+				this.subscription = null;
+				s.cancel();
+			}
+		}
+	}
+
 
 	private class DownstreamBridge implements Subscriber<Void> {
 

spring-web/src/main/java/org/springframework/web/bind/support/WebExchangeDataBinder.java

@@ -72,7 +72,7 @@ public class WebExchangeDataBinder extends WebDataBinder {
 				.map(this::getParamsToBind)
 				.doOnNext(values -> values.putAll(getMultipartFiles(exchange)))
 				.doOnNext(values -> values.putAll(getExtraValuesToBind(exchange)))
-				.then(values -> {
+				.flatMap(values -> {
 					doBind(new MutablePropertyValues(values));
 					return Mono.empty();
 				});

spring-web/src/main/java/org/springframework/web/filter/reactive/HiddenHttpMethodFilter.java

@@ -81,7 +81,7 @@ public class HiddenHttpMethodFilter implements WebFilter {
 					String method = formData.getFirst(this.methodParamName);
 					return StringUtils.hasLength(method) ? mapExchange(exchange, method) : exchange;
 				})
-				.then((exchange1) -> chain.filter(exchange1));
+				.flatMap((exchange1) -> chain.filter(exchange1));
 	}
 
 	private ServerWebExchange mapExchange(ServerWebExchange exchange, String methodParamValue) {

spring-web/src/main/java/org/springframework/web/server/session/DefaultWebSessionManager.java

@@ -104,7 +104,7 @@ public class DefaultWebSessionManager implements WebSessionManager {
 				Flux.fromIterable(getSessionIdResolver().resolveSessionIds(exchange))
 						.concatMap(this.sessionStore::retrieveSession)
 						.next()
-						.then(session -> validateSession(exchange, session))
+						.flatMap(session -> validateSession(exchange, session))
 						.otherwiseIfEmpty(createSession(exchange))
 						.map(session -> extendSession(exchange, session)));
 	}