EventLoopProcessor extends FluxProcessor<IN, IN> implements Runnable { public final boolean awaitAndShutdown() { return awaitAndShutdown(Duration.ofSeconds(-1)); } EventLoopProcessor(
			int bufferSize,
			@Nullable ThreadFactory threadFactory,
			@Nullable ExecutorService executor,
			ExecutorService requestExecutor,
			boolean autoCancel,
			boolean multiproducers,
			Supplier<Slot<IN>> factory,
			WaitStrategy strategy); abstract long getPending(); @Override @Nullable Object scanUnsafe(Attr key); final boolean alive(); final boolean awaitAndShutdown(); @Deprecated final boolean awaitAndShutdown(long timeout, TimeUnit timeUnit); final boolean awaitAndShutdown(Duration timeout); @Override Stream<? extends Scannable> inners(); Flux<IN> drain(); final Flux<IN> forceShutdown(); final long getAvailableCapacity(); @Override @Nullable final Throwable getError(); @Override final String toString(); @Override final int hashCode(); @Override boolean isSerialized(); @Override final boolean isTerminated(); @Override final void onComplete(); @Override final void onError(Throwable t); @Override final void onNext(IN o); @Override final void onSubscribe(final Subscription s); final void shutdown(); @Override final int getBufferSize(); @Override long downstreamCount();  }
@Test public void awaitTerminationImmediate() { assertThat(test.executor).isNotNull(); test.executor.submit(() -> { try { Thread.sleep(100); } catch (InterruptedException e) { e.printStackTrace(); } }); boolean shutResult = test.awaitAndShutdown(Duration.ofMillis(-1)); assertThat(test.executor).isNotNull() .is(shutdown) .isNot(terminated); assertThat(shutResult).isFalse(); }
@Test public void awaitTerminationNanosDuration() { assertThat(test.executor).isNotNull(); test.executor.submit(() -> { try { Thread.sleep(100); } catch (InterruptedException e) { e.printStackTrace(); } }); boolean shutResult = test.awaitAndShutdown(Duration.ofNanos(1000)); assertThat(test.executor).isNotNull() .is(shutdown) .isNot(terminated); assertThat(shutResult).isFalse(); }
@Test public void awaitTerminationNanosLong() { assertThat(test.executor).isNotNull(); test.executor.submit(() -> { try { Thread.sleep(100); } catch (InterruptedException e) { e.printStackTrace(); } }); @SuppressWarnings("deprecation") boolean shutResult = test.awaitAndShutdown(1000, TimeUnit.NANOSECONDS); assertThat(test.executor).isNotNull() .is(shutdown) .isNot(terminated); assertThat(shutResult).isFalse(); }
@Test public void awaitAndShutdownInterrupt() throws InterruptedException { ExecutorService executor = Mockito.mock(ExecutorService.class); Mockito.doThrow(new InterruptedException("boom")) .when(executor) .awaitTermination(ArgumentMatchers.anyLong(), ArgumentMatchers.any()); EventLoopProcessor<String> interruptingProcessor = initProcessor(executor); boolean result = interruptingProcessor.awaitAndShutdown(Duration.ofMillis(100)); assertThat(Thread.currentThread() .isInterrupted()).as("interrupted") .isTrue(); assertThat(result).as("await failed") .isFalse(); }
@Test public void awaitAndShutdownLongInterrupt() throws InterruptedException { ExecutorService executor = Mockito.mock(ExecutorService.class); Mockito.doThrow(new InterruptedException("boom")) .when(executor) .awaitTermination(ArgumentMatchers.anyLong(), ArgumentMatchers.any()); EventLoopProcessor<String> interruptingProcessor = initProcessor(executor); @SuppressWarnings("deprecation") boolean result = interruptingProcessor.awaitAndShutdown(100, TimeUnit.MILLISECONDS); assertThat(Thread.currentThread() .isInterrupted()).as("interrupted") .isTrue(); assertThat(result).as("await failed") .isFalse(); }