/*
 * 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
 *
 *      https://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
 */
@SuppressWarnings("deprecation")
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);
	}

}

@@ -20,10 +20,10 @@
 2-20
-18
+20
  the
@@ -673,275 +673,242 @@
 ort 
-java.time.Duration;%0Aimport java.util.concurrent.TimeUnit;%0A%0Aimport reactor.core.publisher.Mono;%0Aimport reactor.core.publisher.MonoProcessor;%0A%0Aimport org.springframework.lang.Nullable;%0Aimport org.springframework.util.Assert;%0A%0A/**%0A * Adapts a %7B@link Mono%7D into a %7B@link 
+reactor.core.publisher.Mono;%0A%0A/**%0A * Adapts a %7B@link Mono%7D into a %7B@link ListenableFuture%7D by obtaining a%0A * %7B@code CompletableFuture%7D from the %7B@code Mono%7D via %7B@link Mono#toFuture()%7D%0A * and then adapting it with %7B@link CompletableTo
 List
@@ -919,16 +919,23 @@
 leFuture
+Adapter
 %7D.%0A *%0A *
@@ -1041,41 +1041,8 @@
  */%0A
-@SuppressWarnings(%22deprecation%22)%0A
 publ
@@ -1087,1492 +1087,136 @@
 %3CT%3E 
-implements ListenableFuture%3CT%3E %7B%0A%0A%09private final MonoProcessor%3CT%3E processor;%0A%0A%09private final ListenableFutureCallbackRegistry%3CT%3E registry = new ListenableFutureCallbackRegistry%3C%3E();%0A%0A%0A%09public MonoToListenableFutureAdapter(Mono%3CT%3E mono) %7B%0A%09%09Assert.notNull(mono, %22Mono must not be null%22);%0A%09%09this.processor = mono%0A%09%09%09%09.doOnSuccess(this.registry::success)%0A%09%09%09%09.doOnError(this.registry::failure)%0A%09%09%09%09.toProcessor();%0A%09%7D%0A%0A%0A%09@Override%0A%09@Nullable%0A%09public T get() %7B%0A%09%09return this.processor.block();%0A%09%7D%0A%0A%09@Override%0A%09@Nullable%0A%09public T get(long timeout, TimeUnit unit) %7B%0A%09%09Assert.notNull(unit, %22TimeUnit must not be null%22);%0A%09%09Duration duration = Duration.ofMillis(TimeUnit.MILLISECONDS.convert(timeout, unit));%0A%09%09return this.processor.block(duration);%0A%09%7D%0A%0A%09@Override%0A%09public boolean cancel(boolean mayInterruptIfRunning) %7B%0A%09%09if (isCancelled()) %7B%0A%09%09%09return false;%0A%09%09%7D%0A%09%09this.processor.cancel();%0A%09%09// isCancelled may still return false, if mono completed before the cancel%0A%09%09return this.processor.isCancelled();%0A%09%7D%0A%0A%09@Override%0A%09public boolean isCancelled() %7B%0A%09%09return this.processor.isCancelled();%0A%09%7D%0A%0A%09@Override%0A%09public boolean isDone() %7B%0A%09%09return this.processor.isTerminated();%0A%09%7D%0A%0A%09@Override%0A%09public void addCallback(ListenableFutureCallback%3C? super T%3E callback) %7B%0A%09%09this.registry.addCallback(callback);%0A%09%7D%0A%0A%09@Override%0A%09public void addCallback(SuccessCallback%3C? super T%3E success, FailureCallback failure) %7B%0A%09%09this.registry.addSuccessCallback(success);%0A%09%09this.registry.addFailureCallback(failure
+extends CompletableToListenableFutureAdapter%3CT%3E %7B%0A%0A%09public MonoToListenableFutureAdapter(Mono%3CT%3E mono) %7B%0A%09%09super(mono.toFuture()
 );%0A%09

Refactor MonoToListenableFutureAdapter