Move MonoToListenableFutureAdapter to spring-core

This was a package private class in spring-messaging since 5.0, and was
recently made public in 5.1. This commit promotes it to spring-core
where it belongs next to all other ListenableFuture support classes.

Follow-up refactoring for SPR-17336

Author: rstoyanchev

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

@@ -0,0 +1,97 @@
+/*
+ * Copyright 2002-2018 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.util.concurrent;
+
+import java.time.Duration;
+import java.util.concurrent.TimeUnit;
+
+import reactor.core.publisher.Mono;
+import reactor.core.publisher.MonoProcessor;
+
+import org.springframework.lang.Nullable;
+import org.springframework.util.Assert;
+
+/**
+ * Adapts a {@link Mono} into a {@link ListenableFuture}.
+ *
+ * @author Rossen Stoyanchev
+ * @author Stephane Maldini
+ * @since 5.1
+ * @param <T> the object type
+ */
+public class MonoToListenableFutureAdapter<T> implements ListenableFuture<T> {
+
+	private final MonoProcessor<T> processor;
+
+	private final ListenableFutureCallbackRegistry<T> registry = new ListenableFutureCallbackRegistry<>();
+
+
+	public MonoToListenableFutureAdapter(Mono<T> mono) {
+		Assert.notNull(mono, "Mono must not be null");
+		this.processor = mono
+				.doOnSuccess(this.registry::success)
+				.doOnError(this.registry::failure)
+				.toProcessor();
+	}
+
+
+	@Override
+	@Nullable
+	public T get() {
+		return this.processor.block();
+	}
+
+	@Override
+	@Nullable
+	public T get(long timeout, TimeUnit unit) {
+		Assert.notNull(unit, "TimeUnit must not be null");
+		Duration duration = Duration.ofMillis(TimeUnit.MILLISECONDS.convert(timeout, unit));
+		return this.processor.block(duration);
+	}
+
+	@Override
+	public boolean cancel(boolean mayInterruptIfRunning) {
+		if (isCancelled()) {
+			return false;
+		}
+		this.processor.cancel();
+		// isCancelled may still return false, if mono completed before the cancel
+		return this.processor.isCancelled();
+	}
+
+	@Override
+	public boolean isCancelled() {
+		return this.processor.isCancelled();
+	}
+
+	@Override
+	public boolean isDone() {
+		return this.processor.isTerminated();
+	}
+
+	@Override
+	public void addCallback(ListenableFutureCallback<? super T> callback) {
+		this.registry.addCallback(callback);
+	}
+
+	@Override
+	public void addCallback(SuccessCallback<? super T> success, FailureCallback failure) {
+		this.registry.addSuccessCallback(success);
+		this.registry.addFailureCallback(failure);
+	}
+
+}

spring-messaging/src/test/java/org/springframework/messaging/support/MonoToListenableFutureAdapterTests.java renamed to spring-core/src/test/java/org/springframework/util/concurrent/MonoToListenableFutureAdapterTests.java

@@ -13,7 +13,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.springframework.messaging.support;
+package org.springframework.util.concurrent;
 
 import java.time.Duration;
 import java.util.concurrent.Future;
@@ -22,8 +22,6 @@
 import org.junit.Test;
 import reactor.core.publisher.Mono;
 
-import org.springframework.util.concurrent.ListenableFuture;
-
 import static org.junit.Assert.*;
 
 /**

spring-messaging/src/main/java/org/springframework/messaging/handler/invocation/ReactiveReturnValueHandler.java

@@ -21,9 +21,9 @@
 import org.springframework.core.MethodParameter;
 import org.springframework.core.ReactiveAdapter;
 import org.springframework.core.ReactiveAdapterRegistry;
-import org.springframework.messaging.support.MonoToListenableFutureAdapter;
 import org.springframework.util.Assert;
 import org.springframework.util.concurrent.ListenableFuture;
+import org.springframework.util.concurrent.MonoToListenableFutureAdapter;
 
 /**
  * Support for single-value reactive types (like {@code Mono} or {@code Single})

spring-messaging/src/main/java/org/springframework/messaging/support/MonoToListenableFutureAdapter.java

@@ -49,13 +49,13 @@
 
 import org.springframework.lang.Nullable;
 import org.springframework.messaging.Message;
-import org.springframework.messaging.support.MonoToListenableFutureAdapter;
 import org.springframework.messaging.tcp.ReconnectStrategy;
 import org.springframework.messaging.tcp.TcpConnection;
 import org.springframework.messaging.tcp.TcpConnectionHandler;
 import org.springframework.messaging.tcp.TcpOperations;
 import org.springframework.util.Assert;
 import org.springframework.util.concurrent.ListenableFuture;
+import org.springframework.util.concurrent.MonoToListenableFutureAdapter;
 import org.springframework.util.concurrent.SettableListenableFuture;
 
 /**

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

@@ -23,9 +23,9 @@
 import reactor.netty.NettyOutbound;
 
 import org.springframework.messaging.Message;
-import org.springframework.messaging.support.MonoToListenableFutureAdapter;
 import org.springframework.messaging.tcp.TcpConnection;
 import org.springframework.util.concurrent.ListenableFuture;
+import org.springframework.util.concurrent.MonoToListenableFutureAdapter;
 
 /**
  * Reactor Netty based implementation of {@link TcpConnection}.