Issue ReactiveX#13 First draft to publish events as a reactive stream.

Author: Robert Winkler

README.adoc

@@ -4,13 +4,32 @@
 
 image:https://travis-ci.org/javaslang/javaslang-circuitbreaker.svg?branch=master["Build Status", link="https://travis-ci.org/javaslang/javaslang-circuitbreaker"] image:https://coveralls.io/repos/javaslang/javaslang-circuitbreaker/badge.svg["Coverage Status", link="https://coveralls.io/r/javaslang/javaslang-circuitbreaker"] image:https://api.bintray.com/packages/robwin/maven/javaslang-circuitbreaker/images/download.svg[link="https://bintray.com/robwin/maven/javaslang-circuitbreaker/_latestVersion"] image:http://img.shields.io/badge/license-ASF2-blue.svg["Apache License 2", link="http://www.apache.org/licenses/LICENSE-2.0.txt"]
 
-This library is a lightweight, easy-to-use fault tolerance library inspired by https://github.com/Netflix/Hystrix[Netflix Hystrix], but designed solely for Java 8 and functional programming. Lightweight, because the library only uses SLF4J and a functional library for Java 8 called https://github.com/javaslang/javaslang[Javaslang]. Javaslang itself has no external library dependencies. 
-The library provides several higher-order functions to decorate any function with a http://martinfowler.com/bliki/CircuitBreaker.html[Circuit Breaker]. The basic idea behind a CircuitBreaker is simple. You decorate a protected function (e.g. a remote call to a backend system) in a CircuitBreaker, which monitors for failures. Once the failures reach a certain failure threshold, the CircuitBreaker trips, and all further calls return with an exception, without the protected call being made at all. After a certain time duration, the CircuitBreaker allows calls again to see if the backend system is still unreliable or has become available again. 
-In the following I call the higher-order functions decorators. The decorators return an enhanced version of your function. Furthermore, the library provides a decorator to retry failed functions. You can stack more than one decorator on any given function. That means, you can combine a Retry decorator with a CircuitBreaker decorator. Any decorated function can be invoked synchronously or asynchronously.
+This library is a lightweight, easy-to-use fault tolerance library inspired by https://github.com/Netflix/Hystrix[Netflix Hystrix], but designed solely for Java 8 and functional programming. Lightweight, because the library only uses https://github.com/javaslang/javaslang[Javaslang], https://github.com/ReactiveX/RxJava[RxJava] and SLF4J-API. These libraries do no have any other external library dependencies.
+
+To highlight a few differences to Netflix Hystrix:
+
+Executable logic can be passed as simple functions, lambda expressions or method references. In Hystrix, executable logic needs to be wrapped inside a HystrixCommand`.
+
+This library provides higher-order functions to decorate any function, lambda expression or method reference with a http://martinfowler.com/bliki/CircuitBreaker.html[Circuit Breaker]. In the following I call the higher-order functions decorators. The decorators return an enhanced version of your function. Furthermore, the library provides a decorator to retry failed functions. You can stack more than one decorator on any given function. That means, you can combine a Retry decorator with a CircuitBreaker decorator. Any decorated function can be invoked synchronously or asynchronously.
+
+Example:
+
+[source,java]
+----
+CircuitBreaker circuitBreaker = CircuitBreakerRegistry.ofDefaults()
+        .circuitBreaker("backendName");
+Retry retryContext = Retry.ofDefaults();
+
+Try.CheckedSupplier<String> supplier = Retry.decorateCheckedSupplier(retryContext,
+        CircuitBreaker.decorateCheckedSupplier(circuitBreaker, helloWorldService::returnHelloWorld));
+Try<String> result = Try.of(supplier)
+    .recover(throwable -> "Hello from Recovery");
+----
+
 
 The CircuitBreaker is implemented via a finite state machine with three states: `CLOSED`, `OPEN` and `HALF_OPEN`.
 
-image::src/docs/asciidoc/images/state_machine.jpg[]
+image::images/state_machine.jpg[]
 
 The CircuitBreaker does not know anything about the backend's state by itself, but uses the information provided by the decorators via `CircuitBreaker::recordSuccess()` and `CircuitBreaker::recordFailure(throwable)`. See example:
 
@@ -36,15 +55,14 @@ Then, all access to the backend is blocked for a (configurable) time duration. `
 
 The CircuitBreaker uses a Ring Bit Buffer in the `CLOSED` state to store the success or failure statuses of the calls. A successful call is stored as a `0` bit and a failed call is stored as a `1` bit. The Ring Bit Buffer has a (configurable) fixed-size. The Ring Bit Buffer uses internally a https://docs.oracle.com/javase/8/docs/api/java/util/BitSet.html[BitSet] to store the bits which is saving memory compared to a boolean array. The BitSet uses a long[] array to store the bits. That means the BitSet only needs an array of 16 long (64-bit) values to store the status of 1024 calls.
 
-image::src/docs/asciidoc/images/ring_buffer.jpg[Ring Bit Buffer]
+image::images/ring_buffer.jpg[Ring Bit Buffer]
 
 The Ring Bit Buffer must be full, before the failure rate can be calculated.
 For example, if the size of the Ring Buffer is 10, then at least 10 calls must evaluated, before the failure rate can be calculated. If only 9 calls have been evaluated the CircuitBreaker will not trip open even if all 9 calls have failed.
 
 After the time duration has elapsed, the CircuitBreaker state changes from `OPEN` to `HALF_OPEN` and allows calls to see if the backend is still unavailable or has become available again. The CircuitBreaker uses another (configurable) Ring Bit Buffer to evaluate the failure rate in the `HALF_OPEN` state. If the failure rate is above the configured threshold, the state changes back to `OPEN`. If the failure rate is below or equal to the threshold, the state changes back to `CLOSED`.
 `CircuitBreaker::recordFailure(exception)` checks if the exception should be recorded as a failure or should be ignored. You can configure a custom `Predicate` which decides whether an exception should be recorded as a failure. The default Predicate records all exceptions as a failure.
 
-
 == Usage guide
 
 See http://javaslang.github.io/javaslang-circuitbreaker/0.5.0/[User Guide].

src/docs/asciidoc/introduction.adoc

@@ -1,7 +1,27 @@
 == Introduction
 
-This library is a lightweight, easy-to-use fault tolerance library inspired by https://github.com/Netflix/Hystrix[Netflix Hystrix], but designed solely for Java 8 and functional programming. Lightweight, because the library only uses SLF4J and a functional library for Java 8 called https://github.com/javaslang/javaslang[Javaslang]. Javaslang itself has no external library dependencies.
-The library provides several higher-order functions to decorate any function with a http://martinfowler.com/bliki/CircuitBreaker.html[Circuit Breaker]. In the following I call the higher-order functions decorators. The decorators return an enhanced version of your function. Furthermore, the library provides a decorator to retry failed functions. You can stack more than one decorator on any given function. That means, you can combine a Retry decorator with a CircuitBreaker decorator. Any decorated function can be invoked synchronously or asynchronously.
+This library is a lightweight, easy-to-use fault tolerance library inspired by https://github.com/Netflix/Hystrix[Netflix Hystrix], but designed solely for Java 8 and functional programming. Lightweight, because the library only uses https://github.com/javaslang/javaslang[Javaslang], https://github.com/ReactiveX/RxJava[RxJava] and SLF4J-API. These libraries do no have any other external library dependencies.
+
+To highlight a few differences to Netflix Hystrix:
+
+Executable logic can be passed as simple functions, lambda expressions or method references. In Hystrix, executable logic needs to be wrapped inside a HystrixCommand`.
+
+This library provides higher-order functions to decorate any function, lambda expression or method reference with a http://martinfowler.com/bliki/CircuitBreaker.html[Circuit Breaker]. In the following I call the higher-order functions decorators. The decorators return an enhanced version of your function. Furthermore, the library provides a decorator to retry failed functions. You can stack more than one decorator on any given function. That means, you can combine a Retry decorator with a CircuitBreaker decorator. Any decorated function can be invoked synchronously or asynchronously.
+
+Example:
+
+[source,java]
+----
+CircuitBreaker circuitBreaker = CircuitBreakerRegistry.ofDefaults()
+        .circuitBreaker("backendName");
+Retry retryContext = Retry.ofDefaults();
+
+Try.CheckedSupplier<String> supplier = Retry.decorateCheckedSupplier(retryContext,
+        CircuitBreaker.decorateCheckedSupplier(circuitBreaker, helloWorldService::returnHelloWorld));
+Try<String> result = Try.of(supplier)
+    .recover(throwable -> "Hello from Recovery");
+----
+
 
 The CircuitBreaker is implemented via a finite state machine with three states: `CLOSED`, `OPEN` and `HALF_OPEN`.
 

src/docs/asciidoc/usage_guide.adoc

@@ -83,20 +83,18 @@ The following example shows how to ignore an `IOException`, but all other except
 include::../../test/java/javaslang/circuitbreaker/CircuitBreakerTest.java[tags=shouldNotRecordIOExceptionAsAFailure]
 ----
 
-=== Customize the event listener
-The default event listener logs state transitions with INFO level.
-
-----
-INFO  i.g.r.c.i.DefaultCircuitBreakerEventListener - CircuitBreaker 'testName' changes state from CLOSED to OPEN
-----
-
-If you want to use a custom event listener, you have to implement the functional interface `CircuitBreakerEventListener` which has a method `onCircuitBreakerEvent(CircuitBreakerEvent circuitBreakerEvent)`. The only event which currently exists is `CircuitBreakerStateTransitionEvent`.
+=== Observe the event stream
+A CircuitBreaker publishes a stream of CircuitBreakerEvents to any Observer/Consumer who subscribes. This library uses RxJava to to provide this functionality.
+If you want to consume events, you have to subscribe to the event stream. This library provides a consumer `CircuitBreakerEventConsumer` which can be used to store events in a circular buffer with a fixed capacity.
+You can use RxJava to filter certain events.
 
 [source,java]
 ----
-CircuitBreakerConfig circuitBreakerConfig = CircuitBreakerConfig.custom()
-    .onCircuitBreakerEvent((event) -> LOG.info(event.toString()))
-    .build();
+CircuitBreakerEventConsumer ringBuffer = new CircuitBreakerEventConsumer(10);
+CircuitBreaker circuitBreaker = circuitBreakerRegistry.circuitBreaker("testName");
+circuitBreaker.observeCircuitBreakerEvents()
+                .filter(event -> event.getEventType() == Type.RECORDED_FAILURE)
+                .subscribe(ringBuffer);
 ----
 
 === Retry a failed function

src/main/java/javaslang/circuitbreaker/CircuitBreaker.java

@@ -55,36 +55,36 @@ public interface CircuitBreaker {
     void recordSuccess();
 
     /**
-     * Get the name of this CircuitBreaker
+     * Returns the name of this CircuitBreaker
      *
      * @return the name of this CircuitBreaker
      */
     String getName();
 
     /**
-     * Get the state of this CircuitBreaker
+     * Returns the state of this CircuitBreaker
      *
      * @return the state of this CircuitBreaker
      */
     State getState();
 
     /**
-     * Get the CircuitBreakerConfig of this CircuitBreaker.
+     * Returns the CircuitBreakerConfig of this CircuitBreaker.
      *
      * @return the CircuitBreakerConfig of this CircuitBreaker
      */
     CircuitBreakerConfig getCircuitBreakerConfig();
 
     /**
-     * Get the Metrics of this CircuitBreaker.
+     * Returns the Metrics of this CircuitBreaker.
      *
      * @return the Metrics of this CircuitBreaker
      */
     Metrics getMetrics();
 
 
     /**
-     * Get an Observable of CircuitBreakerEvents which can be subscribed
+     * Returns an Observable of CircuitBreakerEvents which can be subscribed
      *
      * @return an Observable of CircuitBreakerEvents which can be subscribed
      */

src/main/java/javaslang/circuitbreaker/CircuitBreakerConfig.java

@@ -18,8 +18,6 @@
  */
 package javaslang.circuitbreaker;
 
-import javaslang.circuitbreaker.internal.DefaultCircuitBreakerEventListener;
-
 import java.time.Duration;
 import java.util.function.Predicate;
 
@@ -30,13 +28,11 @@ public class CircuitBreakerConfig {
     public static final int DEFAULT_WAIT_DURATION_IN_OPEN_STATE = 60; // Seconds
     public static final int DEFAULT_RING_BUFFER_SIZE_IN_HALF_OPEN_STATE = 10;
     public static final int DEFAULT_RING_BUFFER_SIZE_IN_CLOSED_STATE = 100;
-    public static final int DEFAULT_EXCEPTION_RING_BUFFER_SIZE = 10;
 
     private final float failureRateThreshold;
     private final int ringBufferSizeInHalfOpenState;
     private final int ringBufferSizeInClosedState;
     private final Duration waitDurationInOpenState;
-    private final CircuitBreakerEventListener circuitBreakerEventListener;
     private final Predicate<Throwable> recordFailurePredicate;
 
     private CircuitBreakerConfig(Context context){
@@ -45,7 +41,6 @@ private CircuitBreakerConfig(Context context){
         this.ringBufferSizeInHalfOpenState = context.ringBufferSizeInHalfOpenState;
         this.ringBufferSizeInClosedState = context.ringBufferSizeInClosedState;
         this.recordFailurePredicate = context.recordFailurePredicate;
-        this.circuitBreakerEventListener = context.circuitBreakerEventListener;
 
     }
 
@@ -65,10 +60,6 @@ public int getRingBufferSizeInClosedState() {
         return ringBufferSizeInClosedState;
     }
 
-    public CircuitBreakerEventListener getCircuitBreakerEventListener() {
-        return circuitBreakerEventListener;
-    }
-
     public Predicate<Throwable> getRecordFailurePredicate() {
         return recordFailurePredicate;
     }
@@ -162,23 +153,6 @@ public Builder ringBufferSizeInClosedState(int ringBufferSizeInClosedState) {
             return this;
         }
 
-        /**
-         * Deprecated: Better handle events by subscribing to the reactive events stream of a CircuitBreaker instance.
-         *
-         * Configures the CircuitBreakerEventListener which should handle CircuitBreaker events.
-         *
-         * @param circuitBreakerEventListener the CircuitBreakerEventListener which should handle CircuitBreaker events
-         * @return the CircuitBreakerConfig.Builder
-         */
-        @Deprecated
-        public Builder onCircuitBreakerEvent(CircuitBreakerEventListener circuitBreakerEventListener) {
-            if (circuitBreakerEventListener == null) {
-                throw new IllegalArgumentException("circuitBreakerEventListener must not be null");
-            }
-            context.circuitBreakerEventListener = circuitBreakerEventListener;
-            return this;
-        }
-
         /**
          * Configures a Predicate which evaluates if an exception should be recorded as a failure and thus increase the failure rate.
          * The Predicate must return true if the exception should count as a failure, otherwise it must return false.
@@ -206,7 +180,6 @@ private static class Context {
         int ringBufferSizeInHalfOpenState = DEFAULT_RING_BUFFER_SIZE_IN_HALF_OPEN_STATE;
         int ringBufferSizeInClosedState = DEFAULT_RING_BUFFER_SIZE_IN_CLOSED_STATE;
         Duration waitDurationInOpenState = Duration.ofSeconds(DEFAULT_WAIT_DURATION_IN_OPEN_STATE);
-        CircuitBreakerEventListener circuitBreakerEventListener = new DefaultCircuitBreakerEventListener();
         // The default exception predicate counts all exceptions as failures.
         Predicate<Throwable> recordFailurePredicate = (exception) -> true;
     }

src/main/java/javaslang/circuitbreaker/CircuitBreakerEvent.java

@@ -26,7 +26,24 @@ public interface CircuitBreakerEvent {
     /**
      * Returns the name of the CircuitBreaker which has created the event.
      *
-     * @return the name of the CircuitBreaker which has created the event.
+     * @return the name of the CircuitBreaker which has created the event
      */
     String getCircuitBreakerName();
+
+    /**
+     * Returns the type of the CircuitBreaker event.
+     *
+     * @return tthe type of the CircuitBreaker event
+     */
+    Type getEventType();
+
+    /**
+     * Event types which are created by a CircuitBreaker.
+     */
+    enum Type {
+        /** A CircuitBreakerEvent which informs that a failure has been recorded */
+        RECORDED_FAILURE,
+        /** A CircuitBreakerEvent which informs that a failure has been recorded */
+        STATE_TRANSITION
+    }
 }

src/main/java/javaslang/circuitbreaker/CircuitBreakerEventListener.java

@@ -19,7 +19,7 @@
 package javaslang.circuitbreaker;
 
 /**
- * A CircuitBreakerEvent which informs about a state transition.
+ * A CircuitBreakerEvent which informs that a failure has been recorded
  */
 public class CircuitBreakerRecordedFailureEvent implements CircuitBreakerEvent{
 
@@ -36,6 +36,11 @@ public String getCircuitBreakerName() {
         return circuitBreakerName;
     }
 
+    @Override
+    public Type getEventType() {
+        return Type.RECORDED_FAILURE;
+    }
+
     @Override
     public String toString(){
         return String.format("CircuitBreaker '%s' recorded: '%s'", getCircuitBreakerName(), throwable.toString());

src/main/java/javaslang/circuitbreaker/CircuitBreakerRecordedFailureEvent.java

@@ -40,6 +40,11 @@ public String getCircuitBreakerName() {
         return circuitBreakerName;
     }
 
+    @Override
+    public Type getEventType() {
+        return Type.STATE_TRANSITION;
+    }
+
     @Override
     public String toString(){
         return String.format("CircuitBreaker '%s' changes state from %s to %s", getCircuitBreakerName(), getStateTransition().getFromState(), getStateTransition().getToState());