RxUtils { public static <T> Publisher<T> createEmptyPublisher( Supplier<CompletionStage<Void>> supplier ) { return Mono.create( sink -> supplier.get().whenComplete( ( ignore, completionError ) -> { Throwable error = Futures.completionExceptionCause( completionError ); 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 emptyPublisherShouldComplete() throws Throwable { Publisher<Void> emptyPublisher = createEmptyPublisher( Futures::completedWithNull ); StepVerifier.create( emptyPublisher ).verifyComplete(); }
@Test void emptyPublisherShouldErrorIfSupplierErrored() throws Throwable { RuntimeException error = new RuntimeException( "Error" ); Publisher<Void> emptyPublisher = createEmptyPublisher( () -> failedFuture( error ) ); StepVerifier.create( emptyPublisher ).verifyErrorMatches( Predicate.isEqual( error ) ); }