ReactorExt { public static Mono<Optional<Throwable>> emitError(Mono<?> source) { return source.ignoreElement().materialize().map(result -> result.getType() == SignalType.ON_ERROR ? Optional.of(result.getThrowable()) : Optional.empty() ); } private  ReactorExt(); static Function<Mono<T>, Mono<Either<T, Throwable>>> either(); static Mono<Optional<Throwable>> emitError(Mono<?> source); static Flux<T> fromListener(Class<L> listener, Consumer<L> register, Consumer<L> unregister); static Function<Flux<T>, Publisher<T>> badSubscriberHandler(Logger logger); static Function<Flux<List<T>>, Publisher<E>> eventEmitter(
            Function<T, K> keyFun,
            BiPredicate<T, T> valueComparator,
            Function<T, E> valueAddedEventMapper,
            Function<T, E> valueRemovedEventMapper,
            E snapshotEndEvent); static Function<Flux<T>, Publisher<T>> head(Supplier<Collection<T>> headSupplier); static Function<Mono<T>, Mono<T>> hedged(List<Duration> thresholds,
                                                        Predicate<Throwable> retryableErrorPredicate,
                                                        Map<String, String> context,
                                                        Registry registry,
                                                        Scheduler scheduler); static Function<Flux<T>, Publisher<R>> mapWithState(S zero, BiFunction<T, S, Pair<R, S>> transformer); static Function<Flux<T>, Publisher<R>> mapWithState(Supplier<S> zeroSupplier, BiFunction<T, S, Pair<R, S>> transformer); static Function<Flux<T>, Publisher<R>> mapWithState(S zero,
                                                                         BiFunction<T, S, Pair<R, S>> transformer,
                                                                         Flux<Function<S, Pair<R, S>>> cleanupActions); static Mono<Map<K, Optional<Throwable>>> merge(Map<K, Mono<Void>> monos, int concurrencyLimit, Scheduler scheduler); static Function<Flux<T>, Publisher<T>> reEmitter(Function<T, T> transformer, Duration interval, Scheduler scheduler); static void safeDispose(Disposable... disposables); static Mono<T> onWorker(Supplier<T> action, Scheduler.Worker worker); static Mono<T> onWorker(Supplier<T> action, Scheduler.Worker worker, Duration delay); static Mono<Void> onWorker(Runnable action, Scheduler.Worker worker); static Mono<Void> onWorker(Runnable action, Scheduler.Worker worker, Duration delay); @Deprecated static Flux<T> protectFromMissingExceptionHandlers(Flux<T> unprotectedStream, Logger logger); static Flux<T> toFlux(Observable<T> observable); static Mono<T> toMono(Single<T> single); static Mono<Void> toMono(Observable<Void> observable); static Mono<Void> toMono(Completable completable); static Observable<T> toObservable(Flux<T> flux); static Observable<T> toObservable(Mono<T> mono); static Single<T> toSingle(Mono<T> mono); static Completable toCompletable(Mono<Void> mono); static CoreSubscriber<T> silentSubscriber();  }
@Test public void testEmitValue() { Optional<Throwable> error = ReactorExt.emitError(Mono.just("Hello")).block(); assertThat(error).isEmpty(); }
@Test public void testEmitError() { Optional<Throwable> error = ReactorExt.emitError(Mono.error(new RuntimeException("SimulatedError"))).single().block(); assertThat(error).isPresent(); assertThat(error.get()).isInstanceOf(RuntimeException.class); }