package mb.statix.concurrent.actors.impl;

import java.util.Map;
import java.util.concurrent.PriorityBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import javax.annotation.Nullable;

import org.metaborg.util.functions.Function1;
import org.metaborg.util.log.ILogger;
import org.metaborg.util.log.LoggerUtils;

import com.google.common.collect.Maps;

import mb.statix.concurrent.actors.IActor;
import mb.statix.concurrent.actors.IActorMonitor;
import mb.statix.concurrent.actors.IActorRef;
import mb.statix.concurrent.actors.IActorSystem;
import mb.statix.concurrent.actors.TypeTag;

public class ActorSystem implements IActorSystem {

    private static final ILogger logger = LoggerUtils.logger(ActorSystem.class);

    @SuppressWarnings("unused") private static final int PREEMPT_FACTOR = 3;
    @SuppressWarnings("unused") private static final int RESCHEDULE_FACTOR = 7;

    private final Object lock = new Object();
    private final Map<String, Actor<?>> actors;
    private final PriorityBlockingQueue<Runnable> executorQueue;
    private final ThreadPoolExecutor executor;
    private volatile ActorSystemState state;
    private final IActorContext context;


    public ActorSystem() {
        this(Runtime.getRuntime().availableProcessors());
    }

    public ActorSystem(int parallelism) {
        this.actors = Maps.newHashMap();
        this.executorQueue = new PriorityBlockingQueue<>();
        this.executor = new ThreadPoolExecutor(parallelism, parallelism, 60L, TimeUnit.SECONDS, executorQueue);
        this.state = ActorSystemState.INIT;
        this.context = new ActorContext();
    }

    @Override public <T> IActor<T> add(String id, TypeTag<T> type, Function1<IActor<T>, T> supplier) {
        return add(null, id, type, supplier);
    }

    private <T> IActor<T> add(@Nullable IActorRef<?> parent, String id, TypeTag<T> type,
            Function1<IActor<T>, T> supplier) {
        logger.debug("add actor {}", id);
        final Actor<T> actor = new Actor<>(context, id, type, supplier);
        synchronized(lock) {
            if(state.equals(ActorSystemState.STOPPED)) {
                throw new IllegalStateException("Actor system already stopped.");
            }
            if(actors.containsKey(id)) {
                throw new IllegalArgumentException("Actor with id " + id + " already exists.");
            }
            actors.put(id, actor);
            if(state.equals(ActorSystemState.RUNNING)) {
                actor.start();
            }
        }
        logger.debug("added actor {}", id);
        return actor;
    }

    @Override public void addMonitor(IActorRef<?> actor, IActorRef<? extends IActorMonitor> monitor) {
        ((Actor<?>) actor).addMonitor(async(monitor));
    }

    @Override public <T> T async(IActorRef<T> receiver) {
        return ((Actor<T>) receiver).asyncSystem;
    }

    @Override public void start() {
        logger.debug("start system");
        synchronized(lock) {
            if(!state.equals(ActorSystemState.INIT)) {
                throw new IllegalStateException("Actor system already started.");
            }
            state = ActorSystemState.RUNNING;
            for(Actor<?> actor : actors.values()) {
                actor.start();
            }
        }
        logger.debug("started system");
    }

    @Override public void stop() {
        synchronized(lock) {
            if(!state.equals(ActorSystemState.RUNNING)) {
                throw new IllegalStateException("Actor system not started.");
            }
            state = ActorSystemState.STOPPED;
            for(Actor<?> actor : actors.values()) {
                actor.stop();
            }
            executor.shutdown();
        }
    }

    @Override public void cancel() {
        synchronized(lock) {
            if(!state.equals(ActorSystemState.RUNNING)) {
                throw new IllegalStateException("Actor system not started.");
            }
            state = ActorSystemState.STOPPED;
            executor.shutdownNow();
        }
    }

    ///////////////////////////////////////////////////////////////////////////
    // IActorContext
    ///////////////////////////////////////////////////////////////////////////

    @SuppressWarnings({ "unchecked", "rawtypes" })
    private class ActorContext implements IActorContext {

        @Override public <U> IActor<U> add(String id, TypeTag<U> type, Function1<IActor<U>, U> supplier) {
            return ActorSystem.this.add(Actor.current.get(), id, type, supplier);
        }

        @Override public <T> T async(IActorRef<T> receiver) {
            if(!actors.containsValue(receiver)) {
                throw new IllegalArgumentException("Actor " + receiver + " not part of this system.");
            }
            return (T) ((Actor) receiver).asyncActor;
        }

        @Override public ActorTask schedule(Actor<?> actor, int priority) {
            final ActorTask task = new ActorTask(actor, priority);
            executor.execute(task);
            return task;
        }

        @Override public ActorTask reschedule(ActorTask oldTask, int newPriority) {
            if(oldTask.priority * RESCHEDULE_FACTOR < newPriority && executorQueue.remove(oldTask)) {
                return schedule(oldTask.runnable, newPriority);
            } else {
                return oldTask;
            }
        }

        @Override public boolean preempt(int priority) {
            ActorTask top = (ActorTask) executorQueue.peek();
            return top != null && (priority * PREEMPT_FACTOR < top.priority);
        }

    }

    static class ActorTask implements Runnable, Comparable<ActorTask> {

        private final Actor<?> runnable;
        private final int priority;

        ActorTask(Actor<?> runnable, int priority) {
            this.runnable = runnable;
            this.priority = priority;
        }

        @Override public void run() {
            runnable.run();
        }

        @Override public int compareTo(ActorTask o) {
            return priority - o.priority;
        }

    }

}
@@ -6059,16 +6059,18 @@
  return 
+o.
 priority
@@ -6072,18 +6072,16 @@
 ority - 
-o.
 priority

Fix task order.