package fault;

import fault.circuit.BreakerConfig;
import fault.circuit.CircuitBreaker;
import fault.circuit.DefaultCircuitBreaker;
import fault.metrics.ActionMetrics;
import fault.metrics.SingleWriterActionMetrics;

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Created by timbrooks on 12/23/14.
 */
public class BlockingExecutor extends AbstractServiceExecutor implements ServiceExecutor {

    private final ExecutorService service;
    private final ExecutorService managingService = Executors.newFixedThreadPool(2);
    private final DelayQueue<ActionTimeout> timeoutQueue = new DelayQueue<>();
    private final BlockingQueue<ResilientPromise<?>> metricsQueue = new LinkedBlockingQueue<>();

    public BlockingExecutor(int poolSize, int queueSize) {
        this(poolSize, queueSize, new SingleWriterActionMetrics(3600));
    }

    public BlockingExecutor(int poolSize, int queueSize, ActionMetrics actionMetrics) {
        this(poolSize, queueSize, actionMetrics, new DefaultCircuitBreaker(actionMetrics, new BreakerConfig
                .BreakerConfigBuilder
                ().failureThreshold(20).timePeriodInMillis(5000).build()));
    }

    public BlockingExecutor(int poolSize, int queueSize, ActionMetrics actionMetrics, CircuitBreaker circuitBreaker) {
        super(circuitBreaker, actionMetrics);
        this.service = new ThreadPoolExecutor(poolSize, poolSize, Long.MAX_VALUE, TimeUnit.DAYS, new
                ArrayBlockingQueue<Runnable>(queueSize), new ServiceThreadFactory("Hello"));
        startTimeoutAndMetrics();
    }

    @Override
    public <T> ResilientFuture<T> submitAction(final ResilientAction<T> action, long millisTimeout) {
        return submitAction(action, new MultipleWriterResilientPromise<T>(), millisTimeout);
    }

    @Override
    public <T> ResilientFuture<T> submitAction(final ResilientAction<T> action, final ResilientPromise<T> promise,
                                               long millisTimeout) {
        if (!circuitBreaker.allowAction()) {
            throw new RejectedActionException(RejectedActionException.Reason.CIRCUIT_CLOSED);
        }
        final Future<Void> f = service.submit(new Callable<Void>() {
            @Override
            public Void call() {
                try {
                    T result = action.run();
                    if (promise.deliverResult(result)) {
                        metricsQueue.offer(promise);
                    }
                } catch (Exception e) {
                    if (promise.deliverError(e)) {
                        metricsQueue.offer(promise);
                    }
                }
                return null;
            }
        });
        timeoutQueue.offer(new ActionTimeout(promise, millisTimeout, f));
        return new ResilientFuture<>(promise);
    }

    @Override
    public <T> ResilientPromise<T> performAction(final ResilientAction<T> action) {
        ResilientPromise<T> promise = new SingleWriterResilientPromise<>();
        if (!circuitBreaker.allowAction()) {
            throw new RejectedActionException(RejectedActionException.Reason.CIRCUIT_CLOSED);
        }
        try {
            T result = action.run();
            promise.deliverResult(result);
        } catch (ActionTimeoutException e) {
            promise.setTimedOut();
        } catch (Exception e) {
            promise.deliverError(e);
        }
        metricsQueue.offer(promise);
        return promise;
    }

    @Override
    public void shutdown() {
        service.shutdown();
        managingService.shutdown();
        managingService.shutdown();
    }

    private void startTimeoutAndMetrics() {
        managingService.submit(new Runnable() {
            @Override
            public void run() {
                for (; ; ) {
                    try {
                        ResilientPromise<?> promise = metricsQueue.take();
                        actionMetrics.reportActionResult(promise.getStatus());
                        circuitBreaker.informBreakerOfResult(promise.isSuccessful());
                    } catch (InterruptedException e) {
                        break;
                    }

                }
            }
        });

        managingService.submit(new Runnable() {
            @Override
            public void run() {
                for (; ; ) {
                    try {
                        ActionTimeout timeout = timeoutQueue.take();
                        ResilientPromise<?> promise = timeout.promise;
                        if (promise.setTimedOut()) {
                            timeout.future.cancel(true);
                            actionMetrics.reportActionResult(promise.getStatus());
                            circuitBreaker.informBreakerOfResult(promise.isSuccessful());
                        }
                    } catch (InterruptedException e) {
                        break;
                    }
                }
            }
        });
    }

    private class ActionTimeout implements Delayed {

        private final ResilientPromise<?> promise;
        private final long millisAbsoluteTimeout;
        private final Future<Void> future;

        public ActionTimeout(ResilientPromise<?> promise, long millisRelativeTimeout, Future<Void> future) {
            this.promise = promise;
            this.millisAbsoluteTimeout = millisRelativeTimeout + System.currentTimeMillis();
            this.future = future;
        }

        @Override
        public long getDelay(TimeUnit unit) {
            return unit.convert(millisAbsoluteTimeout - System.currentTimeMillis(), TimeUnit.MILLISECONDS);
        }

        @Override
        public int compareTo(Delayed o) {
            return Long.compare(millisAbsoluteTimeout, o.getDelay(TimeUnit.MILLISECONDS));
        }
    }

    private class ServiceThreadFactory implements ThreadFactory {

        private final String name;
        public final AtomicInteger count = new AtomicInteger(0);

        public ServiceThreadFactory(String name) {
            this.name = name;
        }

        @Override
        public Thread newThread(Runnable r) {
            return new Thread(r, name + "-" + count.getAndIncrement());
        }
    }
}

@@ -733,16 +733,47 @@
 eue%3C%3E();
+%0A    private final String name;
 %0A%0A    pu
@@ -858,16 +858,142 @@
 eueSize,
+ null);%0A    %7D%0A%0A    public BlockingExecutor(int poolSize, int queueSize, String name) %7B%0A        this(poolSize, queueSize, name,
  new Sin
@@ -1078,32 +1078,45 @@
 , int queueSize,
+ String name,
  ActionMetrics a
@@ -1164,16 +1164,22 @@
 eueSize,
+ name,
  actionM
@@ -1282,33 +1282,16 @@
 gBuilder
-%0A                
 ().failu
@@ -1405,16 +1405,29 @@
 eueSize,
+ String name,
  ActionM
@@ -1462,16 +1462,28 @@
 tBreaker
+%0A           
  circuit
@@ -1539,16 +1539,142 @@
 trics);%0A
+        if (name == null) %7B%0A            this.name = this.toString();%0A        %7D else %7B%0A            this.name = name;%0A        %7D%0A
         
@@ -1761,20 +1761,16 @@
 it.DAYS,
- new
 %0A       
@@ -1778,16 +1778,20 @@
         
+new 
 ArrayBlo
@@ -1852,15 +1852,8 @@
 ory(
-%22Hello%22
 ));%0A
@@ -6204,43 +6204,8 @@
  %7B%0A%0A
-        private final String name;%0A
     
@@ -6270,100 +6270,8 @@
 );%0A%0A
-        public ServiceThreadFactory(String name) %7B%0A            this.name = name;%0A        %7D%0A%0A
     

Complete implementation of ThreadName