FutureMono extends Mono<Void> { public static <F extends Future<Void>> Mono<Void> from(F future) { Objects.requireNonNull(future, "future"); if(future.isDone()){ if(!future.isSuccess()){ return Mono.error(FutureSubscription.wrapError(future.cause())); } return Mono.empty(); } return new ImmediateFutureMono<>(future); }  static Mono<Void> from(F future); static Mono<Void> deferFuture(Supplier<F> deferredFuture);  }
@Test public void testImmediateFutureMonoImmediate() { ImmediateEventExecutor eventExecutor = ImmediateEventExecutor.INSTANCE; Future<Void> promise = eventExecutor.newFailedFuture(new ClosedChannelException()); StepVerifier.create(FutureMono.from(promise)) .expectError(AbortedException.class) .verify(Duration.ofSeconds(30)); }
@SuppressWarnings("FutureReturnValueIgnored") @Test public void testImmediateFutureMonoLater() { ImmediateEventExecutor eventExecutor = ImmediateEventExecutor.INSTANCE; Promise<Void> promise = eventExecutor.newPromise(); StepVerifier.create(FutureMono.from(promise)) .expectSubscription() .then(() -> promise.setFailure(new ClosedChannelException())) .expectError(AbortedException.class) .verify(Duration.ofSeconds(30)); }