Issue ReactiveX#38: A CircuitBreaker records not permitted calls when the state is OPEN and emits a event for each attempt. The metric "numberOfNotPermittedCalls" can be access via CircuitBreaker.Metrics

Author: Robert Winkler

src/main/java/io/github/robwin/circuitbreaker/CircuitBreaker.java

@@ -225,10 +225,18 @@ interface Metrics {
         /**
          * Returns the current number of failed calls.
          *
-         * @return the current number of failed calls.
+         * @return the current number of failed calls
          */
         int getNumberOfFailedCalls();
 
+        /**
+         * Returns the current number of denied calls.
+         * The number of denied calls is always 0, when the CircuitBreaker state is CLOSED or HALF_OPEN.
+         * The number of denied calls is only increased when the CircuitBreaker state is OPEN.
+         *
+         * @return the current number of denied calls
+         */
+        long getNumberOfNotPermittedCalls();
 
         /**
          * Returns the maximum number of buffered calls.
@@ -237,7 +245,6 @@ interface Metrics {
          */
         int getMaxNumberOfBufferedCalls();
 
-
         /**
          * Returns the maximum number of successful calls.
          *

src/main/java/io/github/robwin/circuitbreaker/event/CircuitBreakerEvent.java

@@ -56,7 +56,9 @@ enum Type {
         IGNORED_ERROR,
         /** A CircuitBreakerEvent which informs that a success has been recorded */
         SUCCESS,
+        /** A CircuitBreakerEvent which informs that a call was not permitted because the CircuitBreaker state is OPEN */
+        NOT_PERMITTED,
         /** A CircuitBreakerEvent which informs the state of the CircuitBreaker has been changed */
-        STATE_TRANSITION
+        STATE_TRANSITION;
     }
 }

src/main/java/io/github/robwin/circuitbreaker/event/CircuitBreakerOnCallNotPermittedEvent.java

@@ -0,0 +1,41 @@
+/*
+ *
+ *  Copyright 2016 Robert Winkler
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *
+ *         http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ *
+ *
+ */
+package io.github.robwin.circuitbreaker.event;
+
+/**
+ * A CircuitBreakerEvent which informs that a call was not permitted, because the CircuitBreaker is OPEN.
+ */
+public class CircuitBreakerOnCallNotPermittedEvent extends AbstractCircuitBreakerEvent {
+
+    public CircuitBreakerOnCallNotPermittedEvent(String circuitBreakerName) {
+        super(circuitBreakerName);
+    }
+
+    @Override
+    public Type getEventType() {
+        return Type.NOT_PERMITTED;
+    }
+
+    @Override
+    public String toString() {
+        return String.format("%s: CircuitBreaker '%s' recorded a call which was not permitted.",
+                getCreationTime(),
+                getCircuitBreakerName());
+    }
+}

src/main/java/io/github/robwin/circuitbreaker/internal/CircuitBreakerMetrics.java

@@ -21,14 +21,18 @@
 
 import io.github.robwin.circuitbreaker.CircuitBreaker;
 
+import java.util.concurrent.atomic.LongAdder;
+
 class CircuitBreakerMetrics implements CircuitBreaker.Metrics {
 
     private final int ringBufferSize;
     private final RingBitSet ringBitSet;
+    private final LongAdder numberOfNotPermittedCalls;
 
     CircuitBreakerMetrics(int ringBufferSize) {
         this.ringBufferSize = ringBufferSize;
         this.ringBitSet = new RingBitSet(this.ringBufferSize);
+        this.numberOfNotPermittedCalls = new LongAdder();
     }
 
     /**
@@ -51,6 +55,12 @@ float onSuccess() {
         return getFailureRate(currentNumberOfFailedCalls);
     }
 
+    /**
+     * Records a call which was not permitted, because the CircuitBreaker state is OPEN.
+     */
+    void onCallNotPermitted() {
+        numberOfNotPermittedCalls.increment();
+    }
 
     /**
      * {@inheritDoc}
@@ -84,6 +94,14 @@ public int getNumberOfBufferedCalls() {
         return this.ringBitSet.length();
     }
 
+    /**
+     * {@inheritDoc}
+     */
+    @Override
+    public long getNumberOfNotPermittedCalls() {
+        return this.numberOfNotPermittedCalls.sum();
+    }
+
     /**
      * {@inheritDoc}
      */

src/main/java/io/github/robwin/circuitbreaker/internal/CircuitBreakerStateMachine.java

@@ -86,7 +86,11 @@ public CircuitBreakerStateMachine(String name, Supplier<CircuitBreakerConfig> ci
      */
     @Override
     public boolean isCallPermitted() {
-        return stateReference.get().isCallPermitted();
+        boolean callPermitted = stateReference.get().isCallPermitted();
+        if(!callPermitted){
+            publishCircuitBreakerEvent(() -> new CircuitBreakerOnCallNotPermittedEvent(getName()));
+        }
+        return callPermitted;
     }
 
     @Override

src/main/java/io/github/robwin/circuitbreaker/internal/OpenState.java

@@ -46,27 +46,30 @@ boolean isCallPermitted() {
             stateMachine.transitionToHalfOpenState();
             return true;
         }
+        circuitBreakerMetrics.onCallNotPermitted();
         return false;
     }
 
     /**
-     * Should never be called, because isCallPermitted returns false.
+     * Should never be called when isCallPermitted returns false.
      */
     @Override
     void onError(Throwable throwable) {
         // Could be called when Thread 1 invokes isCallPermitted when the state is CLOSED, but in the meantime another
-        // Thread 2 calls onError and the state changes from CLOSED to OPEN before Thread 1 calls onError
-        // Do nothing, because the original Exception should be thrown.
+        // Thread 2 calls onError and the state changes from CLOSED to OPEN before Thread 1 calls onError.
+        // But the onError event should still be recorded, even if it happened after the state transition.
+        circuitBreakerMetrics.onError();
     }
 
     /**
-     * Should never be called, because isCallPermitted returns false.
+     * Should never be called when isCallPermitted returns false.
      */
     @Override
     void onSuccess() {
         // Could be called when Thread 1 invokes isCallPermitted when the state is CLOSED, but in the meantime another
-        // Thread 2 calls onError and the state changes from CLOSED to OPEN before Thread 1 calls onSuccess
-        // Do nothing, because the call was successful.
+        // Thread 2 calls onError and the state changes from CLOSED to OPEN before Thread 1 calls onSuccess.
+        // But the onSuccess event should still be recorded, even if it happened after the state transition.
+        circuitBreakerMetrics.onSuccess();
     }
 
     /**

src/test/java/io/github/robwin/circuitbreaker/concurrent/ConcurrentCircuitBreakerTest.java

@@ -54,15 +54,15 @@ public void setUp() {
 
         circuitBreaker = (CircuitBreakerStateMachine) of("testName", circuitBreakerConfig);
         errorEventSubscriber = circuitBreaker.getEventStream()
-                .filter(event -> event.getEventType() == Type.ERROR)
-                .map(CircuitBreakerEvent::getEventType)
-                .test();
+            .filter(event -> event.getEventType() == Type.ERROR)
+            .map(CircuitBreakerEvent::getEventType)
+            .test();
 
         stateTransitionSubsriber = circuitBreaker.getEventStream()
-                .filter(event -> event.getEventType() == Type.STATE_TRANSITION)
-                .cast(CircuitBreakerOnStateTransitionEvent.class)
-                .map(CircuitBreakerOnStateTransitionEvent::getStateTransition)
-                .test();
+            .filter(event -> event.getEventType() == Type.STATE_TRANSITION)
+            .cast(CircuitBreakerOnStateTransitionEvent.class)
+            .map(CircuitBreakerOnStateTransitionEvent::getStateTransition)
+            .test();
     }
 
     @ThreadedMain
@@ -85,6 +85,5 @@ public void arbiter() {
         stateTransitionSubsriber
                 .assertValue(StateTransition.CLOSED_TO_OPEN);
 
-
     }
 }

src/test/java/io/github/robwin/circuitbreaker/internal/CircuitBreakerMetricsTest.java

@@ -33,10 +33,13 @@ public void testDefaultCircuitBreakerMetrics(){
         circuitBreakerMetrics.onSuccess();
         circuitBreakerMetrics.onError();
         circuitBreakerMetrics.onError();
+        circuitBreakerMetrics.onCallNotPermitted();
+        circuitBreakerMetrics.onCallNotPermitted();
 
         assertThat(circuitBreakerMetrics.getNumberOfBufferedCalls()).isEqualTo(4);
         assertThat(circuitBreakerMetrics.getNumberOfFailedCalls()).isEqualTo(2);
         assertThat(circuitBreakerMetrics.getNumberOfSuccessfulCalls()).isEqualTo(2);
+        assertThat(circuitBreakerMetrics.getNumberOfNotPermittedCalls()).isEqualTo(2);
 
         // The failure rate must be -1, because the number of measured calls is below the buffer size of 10
         assertThat(circuitBreakerMetrics.getFailureRate()).isEqualTo(-1);