package fault.java.singlewriter;

import fault.java.ActionMetrics;
import fault.java.ResilientTask;
import fault.java.circuit.CircuitBreaker;

import java.util.HashMap;
import java.util.Map;
import java.util.SortedMap;
import java.util.TreeMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.FutureTask;
import java.util.concurrent.locks.LockSupport;

/**
 * Created by timbrooks on 11/13/14.
 */
public class ManagingRunnable implements Runnable {

    private final CircuitBreaker circuitBreaker;
    private final ActionMetrics actionMetrics;
    private final ConcurrentLinkedQueue<ScheduleMessage<Object>> toScheduleQueue;
    private final ConcurrentLinkedQueue<ResultMessage<Object>> toReturnQueue;
    private final ExecutorService executorService;
    private volatile boolean isRunning;

    public ManagingRunnable(int poolSize, CircuitBreaker circuitBreaker, ActionMetrics actionMetrics) {
        this.circuitBreaker = circuitBreaker;
        this.actionMetrics = actionMetrics;
        this.toScheduleQueue = new ConcurrentLinkedQueue<>();
        this.toReturnQueue = new ConcurrentLinkedQueue<>();
        this.executorService = Executors.newFixedThreadPool(poolSize);
    }

    @Override
    public void run() {
        SortedMap<Long, ResultMessage<Object>> scheduled = new TreeMap<>();
        Map<ResultMessage<Object>, ResilientTask<Object>> taskMap = new HashMap<>();
        isRunning = true;
        while (isRunning) {
            boolean didSomething = false;

            ScheduleMessage<Object> scheduleMessage = toScheduleQueue.poll();
            if (scheduleMessage != null) {
                ActionCallable<Object> actionCallable = new ActionCallable<>(scheduleMessage.action, toReturnQueue);
                FutureTask<Void> futureTask = new FutureTask<>(actionCallable);
                ResilientTask<Object> resilientTask = new ResilientTask<>(futureTask, scheduleMessage.promise);
                ResultMessage<Object> resultMessage = actionCallable.resultMessage;
                taskMap.put(resultMessage, resilientTask);

                executorService.submit(resilientTask);
                scheduled.put(scheduleMessage.relativeTimeout, resultMessage);
                didSomething = true;
            }

            ResultMessage<Object> result = toReturnQueue.poll();
            if (result != null) {
                handleResult(taskMap, result);
                didSomething = true;
            }

            long now = System.currentTimeMillis();
            SortedMap<Long, ResultMessage<Object>> toCancel = scheduled.headMap(now);
            for (Map.Entry<Long, ResultMessage<Object>> entry : toCancel.entrySet()) {
                handleTimeout(taskMap, entry.getValue());
            }

            SortedMap<Long, ResultMessage<Object>> tailView = scheduled.tailMap(now);
            scheduled = new TreeMap<>(tailView);

            if (!didSomething) {
                LockSupport.parkNanos(1);
            }

        }

    }

    @SuppressWarnings("unchecked")
    public <T> void submit(ScheduleMessage<T> message) {
        toScheduleQueue.offer((ScheduleMessage<Object>) message);
    }

    private void handleResult(Map<ResultMessage<Object>, ResilientTask<Object>> taskMap, ResultMessage<Object> result) {
        ResilientTask<Object> resilientTask = taskMap.remove(result);
        if (resilientTask != null) {

            ResilientPromise<Object> promise = resilientTask.resilientPromise;
            if (result.result != null) {
                promise.deliverResult(result.result);
            } else {
                promise.deliverError(result.exception);
            }
            actionMetrics.logActionResult(promise);
            circuitBreaker.informBreakerOfResult(result.exception == null);
        }
    }

    private void handleTimeout(Map<ResultMessage<Object>, ResilientTask<Object>> taskMap, ResultMessage<Object>
            resultMessage) {
        ResilientTask<Object> task = taskMap.remove(resultMessage);
        if (task != null) {
            ResilientPromise<Object> promise = task.resilientPromise;
            if (!promise.isDone()) {
                promise.setTimedOut();
                task.cancel(true);
                actionMetrics.logActionResult(promise);
                circuitBreaker.informBreakerOfResult(false);
            }
        }
    }

    public void shutdown() {
        isRunning = false;
    }
}

@@ -157,91 +157,9 @@
 til.
-HashMap;%0Aimport java.util.Map;%0Aimport java.util.SortedMap;%0Aimport java.util.TreeMap
+*
 ;%0Aim
@@ -1273,32 +1273,37 @@
 SortedMap%3CLong, 
+List%3C
 ResultMessage%3COb
@@ -1308,16 +1308,17 @@
 Object%3E%3E
+%3E
  schedul
@@ -2170,70 +2170,429 @@
     
-scheduled.put(scheduleMessage.relativeTimeout, resultMessage);
+long relativeTimeout = scheduleMessage.relativeTimeout;%0A                if (scheduled.containsKey(relativeTimeout)) %7B%0A                    scheduled.get(relativeTimeout).add(resultMessage);%0A                %7D else %7B%0A                    List%3CResultMessage%3CObject%3E%3E messages = new ArrayList%3C%3E();%0A                    messages.add(resultMessage);%0A                    scheduled.put(relativeTimeout, messages);%0A%0A                %7D
 %0A   
@@ -2909,32 +2909,37 @@
 SortedMap%3CLong, 
+List%3C
 ResultMessage%3COb
@@ -2944,16 +2944,17 @@
 Object%3E%3E
+%3E
  toCance
@@ -3010,24 +3010,29 @@
 Entry%3CLong, 
+List%3C
 ResultMessag
@@ -3041,16 +3041,17 @@
 Object%3E%3E
+%3E
  entry :
@@ -3094,49 +3094,220 @@
     
-handleTimeout(taskMap, entry.getValue());
+List%3CResultMessage%3CObject%3E%3E toTimeout = entry.getValue();%0A                for (ResultMessage%3CObject%3E messageToTimeout : toTimeout) %7B%0A                    handleTimeout(taskMap, messageToTimeout);%0A                %7D
 %0A   
@@ -3346,16 +3346,21 @@
 p%3CLong, 
+List%3C
 ResultMe
@@ -3373,16 +3373,17 @@
 Object%3E%3E
+%3E
  tailVie

Fix some issues with timing out actions that have same relative timeout