@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"); }
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();  }