RxUtils { public static <T> Publisher<T> createMono( Supplier<CompletionStage<T>> supplier ) { return Mono.create( sink -> supplier.get().whenComplete( ( item, completionError ) -> { Throwable error = Futures.completionExceptionCause( completionError ); if ( item != null ) { sink.success( item ); } if ( error != null ) { sink.error( error ); } else { sink.success(); } } ) ); }  static Publisher<T> createEmptyPublisher( Supplier<CompletionStage<Void>> supplier ); static Publisher<T> createMono( Supplier<CompletionStage<T>> supplier );  }
@Test void monoPublisherShouldCompleteWithValue() throws Throwable { Publisher<String> mono = createMono( () -> CompletableFuture.completedFuture( "One" ) ); StepVerifier.create( mono ).expectNext( "One" ).verifyComplete(); }
@Test void monoPublisherShouldErrorIfSupplierErrored() throws Throwable { RuntimeException error = new RuntimeException( "Error" ); Publisher<String> mono = createMono( () -> failedFuture( error ) ); StepVerifier.create( mono ).verifyErrorMatches( Predicate.isEqual( error ) ); }