@Override
	public <T> Mono<T> run(Mono<T> toRun, Function<Throwable, Mono<T>> fallback) {
		Mono<T> toReturn = toRun.transform(new PolarisCircuitBreakerReactorTransformer<>(invokeHandler));
		if (fallback != null) {
			toReturn = toReturn.onErrorResume(throwable -> {
				if (throwable instanceof CallAbortedException) {
					PolarisCircuitBreakerUtils.reportStatus(consumerAPI, conf, (CallAbortedException) throwable);
				}
				return fallback.apply(throwable);
			});
		}
		return toReturn;
	}
@Test
	public void run() {
		this.reactiveContextRunner.run(context -> {
			ReactivePolarisCircuitBreakerFactory polarisCircuitBreakerFactory = context.getBean(ReactivePolarisCircuitBreakerFactory.class);
			ReactiveCircuitBreaker cb = polarisCircuitBreakerFactory.create(SERVICE_CIRCUIT_BREAKER);

			PolarisCircuitBreakerConfigBuilder.PolarisCircuitBreakerConfiguration configuration =
					polarisCircuitBreakerFactory.configBuilder(SERVICE_CIRCUIT_BREAKER).build();

			polarisCircuitBreakerFactory.configureDefault(id -> configuration);

			assertThat(Mono.just("foobar").transform(cb::run).block()).isEqualTo("foobar");

			assertThat(Mono.error(new RuntimeException("boom")).transform(it -> cb.run(it, t -> Mono.just("fallback")))
					.block()).isEqualTo("fallback");

			assertThat(Flux.just("foobar", "hello world").transform(cb::run).collectList().block())
					.isEqualTo(Arrays.asList("foobar", "hello world"));

			assertThat(Flux.error(new RuntimeException("boom")).transform(it -> cb.run(it, t -> Flux.just("fallback")))
					.collectList().block()).isEqualTo(Collections.singletonList("fallback"));
		});
	}