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();  }
@Test(timeout = 5_000) public void testInlineAction() { List<String> result = new CopyOnWriteArrayList<>(); for (int i = 0; i < 5; i++) { int index = i; reactorSerializedInvoker.submit(Mono.just("#" + index).map(tick -> "#" + index)).subscribe(result::add); } await().until(() -> result.size() == 5); assertThat(result).contains("#0", "#1", "#2", "#3", "#4"); }
@Test(timeout = 5_000) public void testLongAsynchronousAction() { List<String> result = new CopyOnWriteArrayList<>(); for (int i = 0; i < 5; i++) { int index = i; reactorSerializedInvoker.submit(Mono.delay(Duration.ofMillis(10)).map(tick -> "#" + index)).subscribe(result::add); } await().until(() -> result.size() == 5); assertThat(result).contains("#0", "#1", "#2", "#3", "#4"); }
@Test(expected = RuntimeException.class, timeout = 5_000) public void testErrorAction() { reactorSerializedInvoker.submit(Mono.error(new RuntimeException("Simulated error"))).block(); }
@Test(timeout = 5_000) public void testQueueingTimeOut() { List<String> result = new CopyOnWriteArrayList<>(); AtomicReference<Throwable> errorRef = new AtomicReference<>(); reactorSerializedInvoker.submit(Mono.delay(Duration.ofHours(1)).timeout(QUEUEING_TIMEOUT).map(tick -> "First")).subscribe( result::add, errorRef::set ); reactorSerializedInvoker.submit(Mono.just("Second")).subscribe(result::add, errorRef::set); await().until(() -> result.size() == 1 && errorRef.get() != null); assertThat(result.get(0)).isEqualTo("Second"); }
@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); }
@Test(timeout = 5_000) public void testCancellation() { AtomicReference<Object> resultRef = new AtomicReference<>(); Disposable disposable = reactorSerializedInvoker.submit(Mono.delay(Duration.ofHours(1)).map(tick -> "First")) .doOnCancel(() -> resultRef.set("CANCELLED")) .subscribe( resultRef::set, resultRef::set ); disposable.dispose(); await().until(disposable::isDisposed); assertThat(resultRef.get()).isEqualTo("CANCELLED"); }