Futures { public static <V> V blockingGet( CompletionStage<V> stage ) { return blockingGet( stage, Futures::noOpInterruptHandler ); } private  Futures(); @SuppressWarnings( "unchecked" ) static CompletableFuture<T> completedWithNull(); static CompletableFuture<T> completeWithNullIfNoError( CompletableFuture<T> future, Throwable error ); static CompletionStage<T> asCompletionStage( io.netty.util.concurrent.Future<T> future ); static CompletionStage<T> asCompletionStage( io.netty.util.concurrent.Future<T> future, CompletableFuture<T> result ); static CompletableFuture<T> failedFuture( Throwable error ); static V blockingGet( CompletionStage<V> stage ); static V blockingGet( CompletionStage<V> stage, Runnable interruptHandler ); static T getNow( CompletionStage<T> stage ); static Throwable completionExceptionCause( Throwable error ); static CompletionException asCompletionException( Throwable error ); static CompletionException combineErrors( Throwable error1, Throwable error2 ); @SuppressWarnings( "ThrowableNotThrown" ) static CompletableFuture<T> onErrorContinue( CompletableFuture<T> future, Throwable errorRecorder,
            Function<Throwable,? extends CompletionStage<T>> onErrorAction );  }
@Test void shouldFailBlockingGetInEventLoopThread() throws Exception { EventLoopGroup eventExecutor = EventLoopGroupFactory.newEventLoopGroup( 1 ); try { CompletableFuture<String> future = new CompletableFuture<>(); Future<String> result = eventExecutor.submit( () -> Futures.blockingGet( future ) ); ExecutionException e = assertThrows( ExecutionException.class, result::get ); assertThat( e.getCause(), is( blockingOperationInEventLoopError() ) ); } finally { eventExecutor.shutdownGracefully(); } }
@Test void shouldThrowInBlockingGetWhenFutureThrowsUncheckedException() { RuntimeException error = new RuntimeException( "Hello" ); CompletableFuture<String> future = new CompletableFuture<>(); future.completeExceptionally( error ); Exception e = assertThrows( Exception.class, () -> Futures.blockingGet( future ) ); assertEquals( error, e ); }
@Test void shouldThrowInBlockingGetWhenFutureThrowsCheckedException() { IOException error = new IOException( "Hello" ); CompletableFuture<String> future = new CompletableFuture<>(); future.completeExceptionally( error ); Exception e = assertThrows( Exception.class, () -> Futures.blockingGet( future ) ); assertEquals( error, e ); }
@Test void shouldReturnFromBlockingGetWhenFutureCompletes() { CompletableFuture<String> future = new CompletableFuture<>(); future.complete( "Hello" ); assertEquals( "Hello", Futures.blockingGet( future ) ); }
@Test void shouldWaitForFutureInBlockingGetEvenWhenInterrupted() { ExecutorService executor = Executors.newSingleThreadExecutor( daemon( "InterruptThread" ) ); try { CompletableFuture<String> future = new CompletableFuture<>(); Thread.currentThread().interrupt(); executor.submit( () -> { sleep( 1_000 ); future.complete( "Hello" ); } ); assertEquals( "Hello", Futures.blockingGet( future ) ); assertTrue( Thread.currentThread().isInterrupted() ); } finally { Thread.interrupted(); executor.shutdown(); } }
@Test void shouldHandleInterruptsInBlockingGet() { try { CompletableFuture<String> future = new CompletableFuture<>(); Thread.currentThread().interrupt(); Runnable interruptHandler = () -> future.complete( "Hello" ); assertEquals( "Hello", Futures.blockingGet( future, interruptHandler ) ); assertTrue( Thread.currentThread().isInterrupted() ); } finally { Thread.interrupted(); } }