public static <T> Publisher<T> onComplete(Publisher<T> publisher, Supplier<CompletableFuture<Void>> future) {
        return actual -> publisher.subscribe(new CompletionAwareSubscriber<T>() {
            @Override
            protected void doOnSubscribe(Subscription subscription) {
                actual.onSubscribe(subscription);
            }

            @Override
            protected void doOnNext(T message) {
                try {
                    actual.onNext(message);
                } catch (Throwable e) {
                    onError(e);
                }

            }

            @Override
            protected void doOnError(Throwable t) {
                actual.onError(t);
            }

            @Override
            protected void doOnComplete() {
                future.get().whenComplete((aVoid, throwable) -> {
                    if (throwable != null) {
                        actual.onError(throwable);
                    } else {
                        actual.onComplete();
                    }
                });
            }
        });
    }
public static < T > Publisher < T > onComplete ( Publisher < T > publisher , Supplier < CompletableFuture < Void > > future ) { return actual -> publisher . subscribe ( new CompletionAwareSubscriber < T > ( ) { @ Override protected void doOnSubscribe ( Subscription subscription ) { actual . onSubscribe ( subscription ) ; } @ Override protected void doOnNext ( T message ) { try { actual . onNext ( message ) ; } catch ( Throwable e ) { onError ( e ) ; } } @ Override protected void doOnError ( Throwable t ) { actual . onError ( t ) ; } @ Override protected void doOnComplete ( ) { future . get ( ) . whenComplete ( ( aVoid , throwable ) -> { if ( throwable != null ) { actual . onError ( throwable ) ; } else { actual . onComplete ( ) ; } } ) ; } } ) ; }
Allow executing logic on completion of a Publisher.

@param publisher The publisher
@param future    The runnable
@param <T>       The generic type
@return The mapped publisher