private < T > void authenticateExecuteAndProcessResults ( Supplier < String > authenticator , Function < String , CompletableFuture < T > > call , final StreamObserver < T > streamObserver , RequestTag requestTag ) { try { String delegationToken ; delegationToken = authenticator . get ( ) ; CompletableFuture < T > result = call . apply ( delegationToken ) ; result . whenComplete ( ( value , ex ) -> { log . debug ( "result = {}" , value ) ; logAndUntrackRequestTag ( requestTag ) ; if ( ex != null ) { Throwable cause = Exceptions . unwrap ( ex ) ; log . error ( "Controller api failed with error: " , ex ) ; String errorDescription = replyWithStackTraceOnError ? "controllerStackTrace=" + Throwables . getStackTraceAsString ( ex ) : cause . getMessage ( ) ; streamObserver . onError ( Status . INTERNAL . withCause ( cause ) . withDescription ( errorDescription ) . asRuntimeException ( ) ) ; } else if ( value != null ) { streamObserver . onNext ( value ) ; streamObserver . onCompleted ( ) ; } } ) ; } catch ( Exception e ) { log . error ( "Controller api failed with authenticator error" ) ; logAndUntrackRequestTag ( requestTag ) ; streamObserver . onError ( Status . UNAUTHENTICATED . withDescription ( "Authentication failed" ) . asRuntimeException ( ) ) ; } }
Convert responses from CompletableFuture to gRPC s Observer pattern .