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 .