Refactor MonoToListenableFutureAdapter

Closes gh-25561

Author: rstoyanchev

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2018 the original author or authors.
+ * Copyright 2002-2020 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.
@@ -16,83 +16,22 @@
 
 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}.
+ * Adapts a {@link Mono} into a {@link ListenableFuture} by obtaining a
+ * {@code CompletableFuture} from the {@code Mono} via {@link Mono#toFuture()}
+ * and then adapting it with {@link CompletableToListenableFutureAdapter}.
  *
  * @author Rossen Stoyanchev
  * @author Stephane Maldini
  * @since 5.1
  * @param <T> the object type
  */
-@SuppressWarnings("deprecation")
-public class MonoToListenableFutureAdapter<T> implements ListenableFuture<T> {
-
-	private final MonoProcessor<T> processor;
-
-	private final ListenableFutureCallbackRegistry<T> registry = new ListenableFutureCallbackRegistry<>();
-
+public class MonoToListenableFutureAdapter<T> extends CompletableToListenableFutureAdapter<T> {
 
 	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);
+		super(mono.toFuture());
 	}
 
 }