package com.github.blemale.fuse;

import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;

import java.time.Duration;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executors;
import java.util.function.Supplier;

public class CircuitBreaker {
    public static class CircuitBreakerOpenException extends RuntimeException {}

    private final StateMachine stateMachine;
    private final RingBuffer<Event> ringBuffer;

    public CircuitBreaker(Condition condition, Duration cooldown) {
        stateMachine = new StateMachine(condition, cooldown);

        Disruptor<Event> disruptor = new Disruptor<>(Event::new, 1024, Executors.defaultThreadFactory());
        disruptor.handleEventsWith(stateMachine);
        disruptor.start();
        ringBuffer = disruptor.getRingBuffer();
    }

    public <T> CompletableFuture<T> executeAsync(Supplier<CompletableFuture<T>> action) {
        try {
            if (stateMachine.isExecutionAllowed()) {
                return action.get().whenCompleteAsync(this::reportResult);
            } else {
                report(CallStatus.OPEN);
                CompletableFuture<T> f = new CompletableFuture<>();
                f.completeExceptionally(new CircuitBreakerOpenException());
                return f;
            }
        } catch (Throwable ex) {
            report(CallStatus.FAILURE);
            CompletableFuture<T> f = new CompletableFuture<>();
            f.completeExceptionally(ex);
            return f;
        }
    }

    public <T> T execute(Supplier<T> action) {
        try {
            if (stateMachine.isExecutionAllowed()) {
                return action.get();
            } else {
                report(CallStatus.OPEN);
                throw new CircuitBreakerOpenException();
            }
        } catch (Throwable ex) {
            report(CallStatus.FAILURE);
            throw ex;
        }
    }


    private <T> void reportResult(T t, Throwable ex) {
        if (ex != null) {
            report(CallStatus.FAILURE);
        } else {
            report(CallStatus.SUCCESS);
        }
    }

    private void report(CallStatus callStatus) {
        ringBuffer.publishEvent((e, seq, s) -> e.setCallStatus(s), callStatus);
    }
}

@@ -285,18 +285,43 @@
 Breaker 
+implements AutoCloseable 
 %7B%0A
-
     publ
@@ -438,16 +438,62 @@
 achine;%0A
+    private final Disruptor%3CEvent%3E disruptor;%0A
     priv
@@ -668,32 +668,15 @@
 );%0A%0A
+
        
- Disruptor%3CEvent%3E
  dis
@@ -1976,16 +1976,79 @@
 %0A    %7D%0A%0A
+    public void close() %7B%0A        disruptor.shutdown();%0A    %7D%0A%0A
 %0A    pri

Make circuit breaker autocloseable to shutdown disruptor