package com.github.blemale.fuse;

import org.agrona.concurrent.*;

import java.time.Clock;
import java.time.Duration;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
import java.util.function.Supplier;

public class CircuitBreaker implements AutoCloseable {
    public static class CircuitBreakerOpenException extends RuntimeException {
        private static CircuitBreakerOpenException INSTANCE = new CircuitBreakerOpenException();
    }

    private final StateMachine stateMachine;
    private final QueuedPipe<CallStatus> queuedPipe;
    private final AgentRunner agentRunner;

    public CircuitBreaker(Condition condition, Duration cooldown) {
        this(condition, cooldown, Clock.systemDefaultZone(), Optional.empty());
    }

    CircuitBreaker(Condition condition, Duration cooldown, Clock clock, Optional<CountDownLatch> latch) {
        stateMachine = new StateMachine(condition, cooldown, clock, latch);
        queuedPipe = new ManyToOneConcurrentArrayQueue<>(1024);
        agentRunner =
                new AgentRunner(
                        new YieldingIdleStrategy(),
                        t -> {},
                        null,
                        new PipeConsumerAgent<>(queuedPipe, stateMachine)
                );
        AgentRunner.startOnThread(agentRunner);
    }

    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(CircuitBreakerOpenException.INSTANCE);
                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()) {
                T result = action.get();
                report(CallStatus.SUCCESS);
                return result;
            } else {
                report(CallStatus.OPEN);
                throw CircuitBreakerOpenException.INSTANCE;
            }
        } catch (Throwable ex) {
            report(CallStatus.FAILURE);
            throw ex;
        }
    }

    @Override
    public void close() {
        agentRunner.close();
    }

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

    private void report(CallStatus callStatus) {
        while (!queuedPipe.offer(callStatus));
    }
}

@@ -111,16 +111,42 @@
 ration;%0A
+import java.util.Objects;%0A
 import j
@@ -929,24 +929,196 @@
 h%3E latch) %7B%0A
+        Objects.requireNonNull(condition);%0A        Objects.requireNonNull(cooldown);%0A        Objects.requireNonNull(clock);%0A        Objects.requireNonNull(latch);%0A        %0A
         stat

Add preconditions to CircuitBreaker