TimeLimiterOperator implements UnaryOperator<Publisher<T>> { public static <T> TimeLimiterOperator<T> of(TimeLimiter timeLimiter) { return new TimeLimiterOperator<>(timeLimiter); } private  TimeLimiterOperator(TimeLimiter timeLimiter); static TimeLimiterOperator<T> of(TimeLimiter timeLimiter); @Override Publisher<T> apply(Publisher<T> publisher);  }
@Test public void doNotTimeoutUsingMono() { given(timeLimiter.getTimeLimiterConfig()) .willReturn(toConfig(Duration.ofMinutes(1))); given(helloWorldService.returnHelloWorld()) .willReturn("Hello world"); Mono<?> mono = Mono.fromCallable(helloWorldService::returnHelloWorld) .transformDeferred(TimeLimiterOperator.of(timeLimiter)); StepVerifier.create(mono) .expectNextCount(1) .verifyComplete(); then(timeLimiter).should(times(2)) .onSuccess(); }
@Test public void timeoutUsingMono() { given(timeLimiter.getTimeLimiterConfig()) .willReturn(toConfig(Duration.ofMillis(1))); Mono<?> mono = Mono.delay(Duration.ofMinutes(1)) .transformDeferred(TimeLimiterOperator.of(timeLimiter)); StepVerifier.create(mono) .expectError(TimeoutException.class) .verify(Duration.ofMinutes(1)); then(timeLimiter).should() .onError(any(TimeoutException.class)); }
@Test public void timeoutNeverUsingMono() { given(timeLimiter.getTimeLimiterConfig()) .willReturn(toConfig(Duration.ofMillis(1))); Mono<?> flux = Mono.never() .transformDeferred(TimeLimiterOperator.of(timeLimiter)); StepVerifier.create(flux) .expectError(TimeoutException.class) .verify(Duration.ofMinutes(1)); then(timeLimiter).should() .onError(any(TimeoutException.class)); }
@Test public void otherErrorUsingMono() { given(timeLimiter.getTimeLimiterConfig()) .willReturn(toConfig(Duration.ofMinutes(1))); given(helloWorldService.returnHelloWorld()) .willThrow(new Error("BAM!")); Mono<?> mono = Mono.fromCallable(helloWorldService::returnHelloWorld) .transformDeferred(TimeLimiterOperator.of(timeLimiter)); StepVerifier.create(mono) .expectError(Error.class) .verify(Duration.ofMinutes(1)); then(timeLimiter).should() .onError(any(Error.class)); }
@Test public void doNotTimeoutUsingFlux() { given(timeLimiter.getTimeLimiterConfig()) .willReturn(toConfig(Duration.ofMinutes(1))); Flux<?> flux = Flux.interval(Duration.ofMillis(1)) .take(2) .transformDeferred(TimeLimiterOperator.of(timeLimiter)); StepVerifier.create(flux) .expectNextCount(2) .verifyComplete(); then(timeLimiter).should(times(3)) .onSuccess(); }
@Test public void timeoutUsingFlux() { given(timeLimiter.getTimeLimiterConfig()) .willReturn(toConfig(Duration.ofMillis(1))); Flux<?> flux = Flux.interval(Duration.ofSeconds(1)) .transformDeferred(TimeLimiterOperator.of(timeLimiter)); StepVerifier.create(flux) .expectError(TimeoutException.class) .verify(Duration.ofMinutes(1)); then(timeLimiter).should() .onError(any(TimeoutException.class)); }
@Test public void timeoutNeverUsingFlux() { given(timeLimiter.getTimeLimiterConfig()) .willReturn(toConfig(Duration.ofMillis(1))); Flux<?> flux = Flux.never() .transformDeferred(TimeLimiterOperator.of(timeLimiter)); StepVerifier.create(flux) .expectError(TimeoutException.class) .verify(Duration.ofMinutes(1)); then(timeLimiter).should() .onError(any(TimeoutException.class)); }
@Test public void otherErrorUsingFlux() { given(timeLimiter.getTimeLimiterConfig()) .willReturn(toConfig(Duration.ofMinutes(1))); Flux<?> flux = Flux.error(new Error("BAM!")) .transformDeferred(TimeLimiterOperator.of(timeLimiter)); StepVerifier.create(flux) .expectError(Error.class) .verify(Duration.ofMinutes(1)); then(timeLimiter).should() .onError(any(Error.class)); }