@Test(timeout = 5_000) public void testShutdown() { List<Disposable> disposables = new ArrayList<>(); AtomicInteger errors = new AtomicInteger(); for (int i = 0; i < 5; i++) { int index = i; Disposable disposable = reactorSerializedInvoker.submit(Mono.delay(Duration.ofHours(1)).map(tick -> "#" + index)) .doFinally(signalType -> errors.incrementAndGet()) .subscribe( next -> { }, e -> { } ); disposables.add(disposable); } reactorSerializedInvoker.shutdown(Duration.ofSeconds(5)); await().until(() -> disposables.stream().map(Disposable::isDisposed).reduce(true, (a, b) -> a && b)); assertThat(errors.get()).isEqualTo(5); }
public void shutdown(Duration timeout) { if (shutdownFlag) { return; } this.shutdownFlag = true; MonoProcessor<Void> marker = MonoProcessor.create(); worker.schedule(this::drainOnShutdown); worker.schedule(marker::onComplete); try { marker.block(timeout); } catch (Exception ignoreTimeout) { } finally { worker.dispose(); } }
ReactorSerializedInvoker { public void shutdown(Duration timeout) { if (shutdownFlag) { return; } this.shutdownFlag = true; MonoProcessor<Void> marker = MonoProcessor.create(); worker.schedule(this::drainOnShutdown); worker.schedule(marker::onComplete); try { marker.block(timeout); } catch (Exception ignoreTimeout) { } finally { worker.dispose(); } } }
ReactorSerializedInvoker { public void shutdown(Duration timeout) { if (shutdownFlag) { return; } this.shutdownFlag = true; MonoProcessor<Void> marker = MonoProcessor.create(); worker.schedule(this::drainOnShutdown); worker.schedule(marker::onComplete); try { marker.block(timeout); } catch (Exception ignoreTimeout) { } finally { worker.dispose(); } } private  ReactorSerializedInvoker(String name,
                                     int size,
                                     Duration excessiveRunningTime,
                                     Scheduler scheduler,
                                     Registry registry,
                                     Clock clock); }
ReactorSerializedInvoker { public void shutdown(Duration timeout) { if (shutdownFlag) { return; } this.shutdownFlag = true; MonoProcessor<Void> marker = MonoProcessor.create(); worker.schedule(this::drainOnShutdown); worker.schedule(marker::onComplete); try { marker.block(timeout); } catch (Exception ignoreTimeout) { } finally { worker.dispose(); } } private  ReactorSerializedInvoker(String name,
                                     int size,
                                     Duration excessiveRunningTime,
                                     Scheduler scheduler,
                                     Registry registry,
                                     Clock clock); void shutdown(Duration timeout); Mono<T> submit(Mono<T> action); static Builder<T> newBuilder(); }
ReactorSerializedInvoker { public void shutdown(Duration timeout) { if (shutdownFlag) { return; } this.shutdownFlag = true; MonoProcessor<Void> marker = MonoProcessor.create(); worker.schedule(this::drainOnShutdown); worker.schedule(marker::onComplete); try { marker.block(timeout); } catch (Exception ignoreTimeout) { } finally { worker.dispose(); } } private  ReactorSerializedInvoker(String name,
                                     int size,
                                     Duration excessiveRunningTime,
                                     Scheduler scheduler,
                                     Registry registry,
                                     Clock clock); void shutdown(Duration timeout); Mono<T> submit(Mono<T> action); static Builder<T> newBuilder();  }