Update after MonoProcessor deprecation in Reactor

This commit adapts the usage of `MonoProcessor` after deprecations
introduced in reactor/reactor-core#1053

Author: bclozel

spring-core/src/main/java/org/springframework/core/codec/Decoder.java

@@ -23,6 +23,7 @@
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.core.publisher.MonoProcessor;
+import reactor.core.publisher.Sinks;
 
 import org.springframework.core.ResolvableType;
 import org.springframework.core.io.buffer.DataBuffer;
@@ -92,7 +93,7 @@ Mono<T> decodeToMono(Publisher<DataBuffer> inputStream, ResolvableType elementTy
 	default T decode(DataBuffer buffer, ResolvableType targetType,
 			@Nullable MimeType mimeType, @Nullable Map<String, Object> hints) throws DecodingException {
 
-		MonoProcessor<T> processor = MonoProcessor.create();
+		MonoProcessor<T> processor = MonoProcessor.fromSink(Sinks.one());
 		decodeToMono(Mono.just(buffer), targetType, mimeType, hints).subscribeWith(processor);
 
 		Assert.state(processor.isTerminated(), "DataBuffer decoding should have completed.");

spring-core/src/main/java/org/springframework/util/concurrent/MonoToListenableFutureAdapter.java

@@ -33,6 +33,7 @@
  * @since 5.1
  * @param <T> the object type
  */
+@SuppressWarnings("deprecation")
 public class MonoToListenableFutureAdapter<T> implements ListenableFuture<T> {
 
 	private final MonoProcessor<T> processor;

spring-messaging/src/main/java/org/springframework/messaging/rsocket/annotation/support/MessagingRSocket.java

@@ -28,6 +28,7 @@
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.core.publisher.MonoProcessor;
+import reactor.core.publisher.Sinks;
 
 import org.springframework.core.io.buffer.DataBuffer;
 import org.springframework.core.io.buffer.DataBufferUtils;
@@ -162,7 +163,7 @@ private int refCount(DataBuffer dataBuffer) {
 	}
 
 	private Flux<Payload> handleAndReply(Payload firstPayload, FrameType frameType, Flux<Payload> payloads) {
-		MonoProcessor<Flux<Payload>> replyMono = MonoProcessor.create();
+		MonoProcessor<Flux<Payload>> replyMono = MonoProcessor.fromSink(Sinks.one());
 		MessageHeaders headers = createHeaders(firstPayload, frameType, replyMono);
 
 		AtomicBoolean read = new AtomicBoolean();

spring-messaging/src/main/java/org/springframework/messaging/tcp/reactor/ReactorNettyTcpClient.java

@@ -34,6 +34,7 @@
 import org.reactivestreams.Publisher;
 import reactor.core.publisher.Mono;
 import reactor.core.publisher.MonoProcessor;
+import reactor.core.publisher.Sinks;
 import reactor.core.scheduler.Scheduler;
 import reactor.core.scheduler.Schedulers;
 import reactor.netty.Connection;
@@ -204,7 +205,7 @@ public ListenableFuture<Void> connect(TcpConnectionHandler<P> handler, Reconnect
 		}
 
 		// Report first connect to the ListenableFuture
-		MonoProcessor<Void> connectMono = MonoProcessor.create();
+		MonoProcessor<Void> connectMono = MonoProcessor.fromSink(Sinks.one());
 
 		this.tcpClient
 				.handle(new ReactorNettyHandler(handler))
@@ -315,7 +316,7 @@ public Publisher<Void> apply(NettyInbound inbound, NettyOutbound outbound) {
 					logger.debug("Connected to " + conn.address());
 				}
 			});
-			MonoProcessor<Void> completion = MonoProcessor.create();
+			MonoProcessor<Void> completion = MonoProcessor.fromSink(Sinks.one());
 			TcpConnection<P> connection = new ReactorNettyTcpConnection<>(inbound, outbound,  codec, completion);
 			scheduler.schedule(() -> this.connectionHandler.afterConnected(connection));
 

spring-test/src/main/java/org/springframework/mock/http/server/reactive/MockServerHttpResponse.java

@@ -26,6 +26,7 @@
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.core.publisher.MonoProcessor;
+import reactor.core.publisher.Sinks;
 
 import org.springframework.core.io.buffer.DataBuffer;
 import org.springframework.core.io.buffer.DataBufferFactory;
@@ -64,7 +65,7 @@ public MockServerHttpResponse(DataBufferFactory dataBufferFactory) {
 		super(dataBufferFactory);
 		this.writeHandler = body -> {
 			// Avoid .then() which causes data buffers to be released
-			MonoProcessor<Void> completion = MonoProcessor.create();
+			MonoProcessor<Void> completion = MonoProcessor.fromSink(Sinks.one());
 			this.body = body.doOnComplete(completion::onComplete).doOnError(completion::onError).cache();
 			this.body.subscribe();
 			return completion;

spring-test/src/main/java/org/springframework/test/web/reactive/server/HttpHandlerConnector.java

@@ -25,6 +25,7 @@
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.core.publisher.MonoProcessor;
+import reactor.core.publisher.Sinks;
 import reactor.core.scheduler.Schedulers;
 
 import org.springframework.core.io.buffer.DataBuffer;
@@ -83,8 +84,8 @@ public Mono<ClientHttpResponse> connect(HttpMethod httpMethod, URI uri,
 	private Mono<ClientHttpResponse> doConnect(
 			HttpMethod httpMethod, URI uri, Function<? super ClientHttpRequest, Mono<Void>> requestCallback) {
 
-		MonoProcessor<Void> requestWriteCompletion = MonoProcessor.create();
-		MonoProcessor<Void> handlerCompletion = MonoProcessor.create();
+		MonoProcessor<Void> requestWriteCompletion = MonoProcessor.fromSink(Sinks.one());
+		MonoProcessor<Void> handlerCompletion = MonoProcessor.fromSink(Sinks.one());
 		ClientHttpResponse[] savedResponse = new ClientHttpResponse[1];
 
 		MockClientHttpRequest mockClientRequest = new MockClientHttpRequest(httpMethod, uri);

spring-test/src/main/java/org/springframework/test/web/reactive/server/WiretapConnector.java

@@ -27,6 +27,7 @@
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.core.publisher.MonoProcessor;
+import reactor.core.publisher.Sinks;
 
 import org.springframework.core.io.buffer.DataBuffer;
 import org.springframework.core.io.buffer.DefaultDataBufferFactory;
@@ -132,7 +133,7 @@ final static class WiretapRecorder {
 
 		private final DataBuffer buffer = DefaultDataBufferFactory.sharedInstance.allocateBuffer();
 
-		private final MonoProcessor<byte[]> content = MonoProcessor.create();
+		private final MonoProcessor<byte[]> content = MonoProcessor.fromSink(Sinks.one());
 
 		private boolean hasContentConsumer;
 

spring-webflux/src/main/java/org/springframework/web/reactive/result/method/SyncInvocableHandlerMethod.java

@@ -22,6 +22,7 @@
 import java.util.stream.Collectors;
 
 import reactor.core.publisher.MonoProcessor;
+import reactor.core.publisher.Sinks;
 
 import org.springframework.core.DefaultParameterNameDiscoverer;
 import org.springframework.core.ParameterNameDiscoverer;
@@ -102,7 +103,7 @@ public ParameterNameDiscoverer getParameterNameDiscoverer() {
 	public HandlerResult invokeForHandlerResult(ServerWebExchange exchange,
 			BindingContext bindingContext, Object... providedArgs) {
 
-		MonoProcessor<HandlerResult> processor = MonoProcessor.create();
+		MonoProcessor<HandlerResult> processor = MonoProcessor.fromSink(Sinks.one());
 		this.delegate.invoke(exchange, bindingContext, providedArgs).subscribeWith(processor);
 
 		if (processor.isTerminated()) {

spring-webflux/src/main/java/org/springframework/web/reactive/result/method/annotation/ModelAttributeMethodArgumentResolver.java

@@ -25,6 +25,7 @@
 
 import reactor.core.publisher.Mono;
 import reactor.core.publisher.MonoProcessor;
+import reactor.core.publisher.Sinks;
 
 import org.springframework.beans.BeanUtils;
 import org.springframework.core.DefaultParameterNameDiscoverer;
@@ -116,7 +117,7 @@ public Mono<Object> resolveArgument(
 		Mono<?> valueMono = prepareAttributeMono(name, valueType, context, exchange);
 
 		Map<String, Object> model = context.getModel().asMap();
-		MonoProcessor<BindingResult> bindingResultMono = MonoProcessor.create();
+		MonoProcessor<BindingResult> bindingResultMono = MonoProcessor.fromSink(Sinks.one());
 		model.put(BindingResult.MODEL_KEY_PREFIX + name, bindingResultMono);
 
 		return valueMono.flatMap(value -> {

spring-webflux/src/main/java/org/springframework/web/reactive/socket/adapter/AbstractListenerWebSocketSession.java

@@ -26,6 +26,7 @@
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.core.publisher.MonoProcessor;
+import reactor.core.publisher.Sinks;
 import reactor.util.concurrent.Queues;
 
 import org.springframework.core.io.buffer.DataBufferFactory;
@@ -73,7 +74,7 @@ public abstract class AbstractListenerWebSocketSession<T> extends AbstractWebSoc
 
 	private final AtomicBoolean sendCalled = new AtomicBoolean();
 
-	private final MonoProcessor<CloseStatus> closeStatusProcessor = MonoProcessor.create();
+	private final MonoProcessor<CloseStatus> closeStatusProcessor = MonoProcessor.fromSink(Sinks.one());
 
 
 	/**