From 7758ba3c7e4f21435ed927417eb3a13c37af5551 Mon Sep 17 00:00:00 2001 From: Rossen Stoyanchev Date: Fri, 14 Aug 2020 10:16:20 +0100 Subject: [PATCH] Refactor MonoToListenableFutureAdapter Closes gh-25561 --- .../MonoToListenableFutureAdapter.java | 73 ++----------------- 1 file changed, 6 insertions(+), 67 deletions(-) diff --git a/spring-core/src/main/java/org/springframework/util/concurrent/MonoToListenableFutureAdapter.java b/spring-core/src/main/java/org/springframework/util/concurrent/MonoToListenableFutureAdapter.java index 61901abda6..5f58ae1924 100644 --- a/spring-core/src/main/java/org/springframework/util/concurrent/MonoToListenableFutureAdapter.java +++ b/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 the object type */ -@SuppressWarnings("deprecation") -public class MonoToListenableFutureAdapter implements ListenableFuture { - - private final MonoProcessor processor; - - private final ListenableFutureCallbackRegistry registry = new ListenableFutureCallbackRegistry<>(); - +public class MonoToListenableFutureAdapter extends CompletableToListenableFutureAdapter { public MonoToListenableFutureAdapter(Mono 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 callback) { - this.registry.addCallback(callback); - } - - @Override - public void addCallback(SuccessCallback success, FailureCallback failure) { - this.registry.addSuccessCallback(success); - this.registry.addFailureCallback(failure); + super(mono.toFuture()); } }