@Test(timeout = 5_000) public void testExcessiveRunningTime() { AtomicReference<Object> resultRef = new AtomicReference<>(); Disposable disposable = reactorSerializedInvoker.submit(Mono.delay(Duration.ofHours(1)).map(tick -> "First")).subscribe( resultRef::set, resultRef::set ); await().until(disposable::isDisposed); assertThat(resultRef.get()).isInstanceOf(TimeoutException.class); }
public Mono<T> submit(Mono<T> action) { Preconditions.checkState(!shutdownFlag, "ReactorQueue has been shutdown"); Preconditions.checkNotNull(action); StackTraceElement[] stackTrace = Thread.currentThread().getStackTrace(); return Mono.create(sink -> { metrics.onSubmit(); ActionHandler actionHandler = new ActionHandler(action, sink, stackTrace); if (!actionHandlers.offer(actionHandler)) { metrics.onQueueFull(); sink.error(actionHandler.newException(new IllegalStateException("Queue is full"))); return; } metrics.setQueueSize(actionHandlers.size()); if (shutdownFlag) { actionHandler.terminate(); } else { worker.schedule(this::drain); } }); }
ReactorSerializedInvoker { public Mono<T> submit(Mono<T> action) { Preconditions.checkState(!shutdownFlag, "ReactorQueue has been shutdown"); Preconditions.checkNotNull(action); StackTraceElement[] stackTrace = Thread.currentThread().getStackTrace(); return Mono.create(sink -> { metrics.onSubmit(); ActionHandler actionHandler = new ActionHandler(action, sink, stackTrace); if (!actionHandlers.offer(actionHandler)) { metrics.onQueueFull(); sink.error(actionHandler.newException(new IllegalStateException("Queue is full"))); return; } metrics.setQueueSize(actionHandlers.size()); if (shutdownFlag) { actionHandler.terminate(); } else { worker.schedule(this::drain); } }); } }
ReactorSerializedInvoker { public Mono<T> submit(Mono<T> action) { Preconditions.checkState(!shutdownFlag, "ReactorQueue has been shutdown"); Preconditions.checkNotNull(action); StackTraceElement[] stackTrace = Thread.currentThread().getStackTrace(); return Mono.create(sink -> { metrics.onSubmit(); ActionHandler actionHandler = new ActionHandler(action, sink, stackTrace); if (!actionHandlers.offer(actionHandler)) { metrics.onQueueFull(); sink.error(actionHandler.newException(new IllegalStateException("Queue is full"))); return; } metrics.setQueueSize(actionHandlers.size()); if (shutdownFlag) { actionHandler.terminate(); } else { worker.schedule(this::drain); } }); } private  ReactorSerializedInvoker(String name,
                                     int size,
                                     Duration excessiveRunningTime,
                                     Scheduler scheduler,
                                     Registry registry,
                                     Clock clock); }
ReactorSerializedInvoker { public Mono<T> submit(Mono<T> action) { Preconditions.checkState(!shutdownFlag, "ReactorQueue has been shutdown"); Preconditions.checkNotNull(action); StackTraceElement[] stackTrace = Thread.currentThread().getStackTrace(); return Mono.create(sink -> { metrics.onSubmit(); ActionHandler actionHandler = new ActionHandler(action, sink, stackTrace); if (!actionHandlers.offer(actionHandler)) { metrics.onQueueFull(); sink.error(actionHandler.newException(new IllegalStateException("Queue is full"))); return; } metrics.setQueueSize(actionHandlers.size()); if (shutdownFlag) { actionHandler.terminate(); } else { worker.schedule(this::drain); } }); } 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 Mono<T> submit(Mono<T> action) { Preconditions.checkState(!shutdownFlag, "ReactorQueue has been shutdown"); Preconditions.checkNotNull(action); StackTraceElement[] stackTrace = Thread.currentThread().getStackTrace(); return Mono.create(sink -> { metrics.onSubmit(); ActionHandler actionHandler = new ActionHandler(action, sink, stackTrace); if (!actionHandlers.offer(actionHandler)) { metrics.onQueueFull(); sink.error(actionHandler.newException(new IllegalStateException("Queue is full"))); return; } metrics.setQueueSize(actionHandlers.size()); if (shutdownFlag) { actionHandler.terminate(); } else { worker.schedule(this::drain); } }); } 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();  }