@Test void ignoresCancellationPropagation() { NoopRequest request = new NoopRequest(Duration.ZERO, mock(RequestContext.class), mock(RetryStrategy.class), mock(CollectionIdentifier.class)); Mono<NoopResponse> mono = Reactor.wrap(request, request.response(), false); assertThrows(Exception.class, () -> mono.timeout(Duration.ofMillis(10)).block()); assertFalse(request.response().isCompletedExceptionally()); assertFalse(request.response().isDone()); }
public static <T> Mono<T> wrap(final Request<?> request, final CompletableFuture<T> response, final boolean propagateCancellation) { Mono<T> mono = MyLittleAssemblyFactory.callOnAssembly(new SilentMonoCompletionStage<>(response)); if (propagateCancellation) { mono = mono.doFinally(st -> { if (st == SignalType.CANCEL) { request.cancel(CancellationReason.STOPPED_LISTENING); } }); } return mono.onErrorResume(err -> { if (err instanceof CompletionException) { return Mono.error(err.getCause()); } else { return Mono.error(err); } }); }
Reactor { public static <T> Mono<T> wrap(final Request<?> request, final CompletableFuture<T> response, final boolean propagateCancellation) { Mono<T> mono = MyLittleAssemblyFactory.callOnAssembly(new SilentMonoCompletionStage<>(response)); if (propagateCancellation) { mono = mono.doFinally(st -> { if (st == SignalType.CANCEL) { request.cancel(CancellationReason.STOPPED_LISTENING); } }); } return mono.onErrorResume(err -> { if (err instanceof CompletionException) { return Mono.error(err.getCause()); } else { return Mono.error(err); } }); } }
Reactor { public static <T> Mono<T> wrap(final Request<?> request, final CompletableFuture<T> response, final boolean propagateCancellation) { Mono<T> mono = MyLittleAssemblyFactory.callOnAssembly(new SilentMonoCompletionStage<>(response)); if (propagateCancellation) { mono = mono.doFinally(st -> { if (st == SignalType.CANCEL) { request.cancel(CancellationReason.STOPPED_LISTENING); } }); } return mono.onErrorResume(err -> { if (err instanceof CompletionException) { return Mono.error(err.getCause()); } else { return Mono.error(err); } }); } private  Reactor(); }
Reactor { public static <T> Mono<T> wrap(final Request<?> request, final CompletableFuture<T> response, final boolean propagateCancellation) { Mono<T> mono = MyLittleAssemblyFactory.callOnAssembly(new SilentMonoCompletionStage<>(response)); if (propagateCancellation) { mono = mono.doFinally(st -> { if (st == SignalType.CANCEL) { request.cancel(CancellationReason.STOPPED_LISTENING); } }); } return mono.onErrorResume(err -> { if (err instanceof CompletionException) { return Mono.error(err.getCause()); } else { return Mono.error(err); } }); } private  Reactor(); static Mono<T> wrap(final Request<?> request, final CompletableFuture<T> response,
                                 final boolean propagateCancellation); static Mono<T> toMono(Supplier<CompletableFuture<T>> input); static Flux<T> toFlux(Supplier<CompletableFuture<C>> input); }
Reactor { public static <T> Mono<T> wrap(final Request<?> request, final CompletableFuture<T> response, final boolean propagateCancellation) { Mono<T> mono = MyLittleAssemblyFactory.callOnAssembly(new SilentMonoCompletionStage<>(response)); if (propagateCancellation) { mono = mono.doFinally(st -> { if (st == SignalType.CANCEL) { request.cancel(CancellationReason.STOPPED_LISTENING); } }); } return mono.onErrorResume(err -> { if (err instanceof CompletionException) { return Mono.error(err.getCause()); } else { return Mono.error(err); } }); } private  Reactor(); static Mono<T> wrap(final Request<?> request, final CompletableFuture<T> response,
                                 final boolean propagateCancellation); static Mono<T> toMono(Supplier<CompletableFuture<T>> input); static Flux<T> toFlux(Supplier<CompletableFuture<C>> input);  }