ReactiveX#212 Add more tests checking we get the right exception (ReactiveX#214)

Check that we are getting the resilience exception even when the source
will throw an exception (not on subscribe)

Author: madgnome

resilience4j-reactor/src/test/java/io/github/resilience4j/reactor/CombinedOperatorsTest.java

@@ -80,4 +80,16 @@ public void shouldEmitErrorWithCircuitBreakerOpenExceptionEvenWhenErrorDuringSub
         ).expectError(CircuitBreakerOpenException.class)
                 .verify(Duration.ofSeconds(1));
     }
+
+    @Test
+    public void shouldEmitErrorWithCircuitBreakerOpenExceptionEvenWhenErrorNotOnSubscribe() {
+        circuitBreaker.transitionToOpenState();
+        StepVerifier.create(
+                Flux.error(new IOException("BAM!"), true)
+                        .transform(CircuitBreakerOperator.of(circuitBreaker))
+                        .transform(BulkheadOperator.of(bulkhead, Schedulers.immediate()))
+                        .transform(RateLimiterOperator.of(rateLimiter, Schedulers.immediate()))
+        ).expectError(CircuitBreakerOpenException.class)
+                .verify(Duration.ofSeconds(1));
+    }
 }

resilience4j-reactor/src/test/java/io/github/resilience4j/reactor/bulkhead/operator/FluxBulkheadTest.java

@@ -84,4 +84,18 @@ public void shouldEmitBulkheadFullExceptionEvenWhenErrorDuringSubscribe() {
 
         assertThat(bulkhead.getMetrics().getAvailableConcurrentCalls()).isEqualTo(0);
     }
+
+    @Test
+    public void shouldEmitBulkheadFullExceptionEvenWhenErrorNotOnSubscribe() {
+        bulkhead.isCallPermitted();
+
+        StepVerifier.create(
+                Flux.error(new IOException("BAM!"), true)
+                        .transform(BulkheadOperator.of(bulkhead, Schedulers.immediate())))
+                .expectSubscription()
+                .expectError(BulkheadFullException.class)
+                .verify(Duration.ofSeconds(1));
+
+        assertThat(bulkhead.getMetrics().getAvailableConcurrentCalls()).isEqualTo(0);
+    }
 }

resilience4j-reactor/src/test/java/io/github/resilience4j/reactor/bulkhead/operator/MonoBulkheadTest.java

@@ -83,4 +83,18 @@ public void shouldEmitBulkheadFullExceptionEvenWhenErrorDuringSubscribe() {
 
         assertThat(bulkhead.getMetrics().getAvailableConcurrentCalls()).isEqualTo(0);
     }
+
+    @Test
+    public void shouldEmitBulkheadFullExceptionEvenWhenErrorNotOnSubscribe() {
+        bulkhead.isCallPermitted();
+
+        StepVerifier.create(
+                Mono.error(new IOException("BAM!")).delayElement(Duration.ofMillis(1))
+                        .transform(BulkheadOperator.of(bulkhead, Schedulers.immediate())))
+                .expectSubscription()
+                .expectError(BulkheadFullException.class)
+                .verify(Duration.ofSeconds(1));
+
+        assertThat(bulkhead.getMetrics().getAvailableConcurrentCalls()).isEqualTo(0);
+    }
 }

resilience4j-reactor/src/test/java/io/github/resilience4j/reactor/circuitbreaker/operator/MonoCircuitBreakerTest.java

@@ -47,6 +47,18 @@ public void shouldPropagateError() {
         assertSingleFailedCall();
     }
 
+    @Test
+    public void shouldEmitCircuitBreakerOpenExceptionEvenWhenErrorNotOnSubscribe() {
+        circuitBreaker.transitionToForcedOpenState();
+        StepVerifier.create(
+                Mono.error(new IOException("BAM!")).delayElement(Duration.ofMillis(1))
+                        .transform(CircuitBreakerOperator.of(circuitBreaker)))
+                .expectError(CircuitBreakerOpenException.class)
+                .verify(Duration.ofSeconds(1));
+
+        assertNoRegisteredCall();
+    }
+
     @Test
     public void shouldEmitCircuitBreakerOpenExceptionEvenWhenErrorDuringSubscribe() {
         circuitBreaker.transitionToForcedOpenState();

resilience4j-reactor/src/test/java/io/github/resilience4j/reactor/ratelimiter/operator/FluxRateLimiterTest.java

@@ -51,7 +51,7 @@ public void shouldPropagateError() {
     }
 
     @Test
-    public void shouldEmitErrorWithBulkheadFullException() {
+    public void shouldEmitRequestNotPermittedException() {
         saturateRateLimiter();
 
         StepVerifier.create(
@@ -63,4 +63,32 @@ public void shouldEmitErrorWithBulkheadFullException() {
 
         assertNoPermitLeft();
     }
+
+    @Test
+    public void shouldEmitRequestNotPermittedExceptionEvenWhenErrorDuringSubscribe() {
+        saturateRateLimiter();
+
+        StepVerifier.create(
+                Flux.error(new IOException("BAM!"))
+                        .transform(RateLimiterOperator.of(rateLimiter)))
+                .expectSubscription()
+                .expectError(RequestNotPermitted.class)
+                .verify(Duration.ofSeconds(1));
+
+        assertNoPermitLeft();
+    }
+
+    @Test
+    public void shouldEmitRequestNotPermittedExceptionEvenWhenErrorNotOnSubscribe() {
+        saturateRateLimiter();
+
+        StepVerifier.create(
+                Flux.error(new IOException("BAM!"), true)
+                        .transform(RateLimiterOperator.of(rateLimiter)))
+                .expectSubscription()
+                .expectError(RequestNotPermitted.class)
+                .verify(Duration.ofSeconds(1));
+
+        assertNoPermitLeft();
+    }
 }

resilience4j-reactor/src/test/java/io/github/resilience4j/reactor/ratelimiter/operator/MonoRateLimiterTest.java

@@ -64,7 +64,7 @@ public void shouldEmitErrorWithBulkheadFullException() {
     }
 
     @Test
-    public void shouldEmitCircuitBreakerOpenExceptionEvenWhenErrorDuringSubscribe() {
+    public void shouldEmitRequestNotPermittedExceptionEvenWhenErrorDuringSubscribe() {
         saturateRateLimiter();
         StepVerifier.create(
                 Mono.error(new IOException("BAM!"))
@@ -74,4 +74,16 @@ public void shouldEmitCircuitBreakerOpenExceptionEvenWhenErrorDuringSubscribe()
 
         assertNoPermitLeft();
     }
+
+    @Test
+    public void shouldEmitRequestNotPermittedExceptionEvenWhenErrorNotOnSubscribe() {
+        saturateRateLimiter();
+        StepVerifier.create(
+                Mono.error(new IOException("BAM!")).delayElement(Duration.ofMillis(1))
+                        .transform(RateLimiterOperator.of(rateLimiter, Schedulers.immediate())))
+                .expectError(RequestNotPermitted.class)
+                .verify(Duration.ofSeconds(1));
+
+        assertNoPermitLeft();
+    }
 }