package net.ranides.assira.events;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Consumer;
import java.util.function.Supplier;

/**
 * TaskRouter is used to schedule executable tasks and optionally wait for generated result.
 *
 * It can be used in synchronous or asynchronous way: that means,
 * you can wait for results or you can pass callbacks.
 */
public class TaskRouter {

    private static final Object SUCCESS = new Object();

    private static final AtomicLong COUNTER = new AtomicLong(0);

    private final long tid;

    private final AtomicLong cid = new AtomicLong(0);

    private final EventRouter producer;

    private final EventRouter consumer;

    private final ConcurrentHashMap<Long, Consumer<Object>> callbacks = new ConcurrentHashMap<>();

    /**
     * Creates new TaskRouter, which uses default EventProactor as task scheduler.
     * It means: it uses the same task scheduler to transfer events with requests and responses.
     *
     * @param name name
     */
    public TaskRouter(String name) {
        this(EventProactor.newInstance(name));
    }

    /**
     * Creates new TaskRouter, which uses provided EventRouter as task scheduler.
     * It means: it uses the same task scheduler to transfer events with requests and responses.
     *
     * @param router router
     */
    public TaskRouter(EventRouter router) {
        this(router, router);
    }

    /**
     * Creates new TaskRouter, which uses provided EventRouters for scheduling tasks.
     * "requests" is used to store events about peding requests.
     * "responses" is used to store events about finished tasks.
     *
     * @param requests requests
     * @param responses responses
     */
    public TaskRouter(EventRouter requests, EventRouter responses) {
        this.tid = COUNTER.incrementAndGet();
        this.producer = requests;
        this.consumer = responses;

        this.producer.addEventListener(RunnableEvent.class, event -> {
            if(event.tid == tid) {
                event.action.run();
            }
        });

        this.producer.addEventListener(SupplierEvent.class, event -> {
            if(event.tid == tid) {
                responses.signalEvent(new SupplierResponse(tid, event.cid, event.function.get()));
            }
        });

        this.consumer.addEventListener(SupplierResponse.class, event -> {
            if(event.tid == tid) {
                Consumer<Object> callback = callbacks.remove(event.cid);
                if(callback != null) {
                    callback.accept(event.value);
                }
            }
        });
    }

    /**
     * Schedules executable action.
     * Does not wait for finishing the job.
     *
     * @param action action
     */
    public void schedule(Runnable action) {
        producer.signalEvent(new RunnableEvent(tid, action));
    }

    /**
     * Schedules executable action.
     * Executes "callback" after "action" finished work.
     *
     * Does not block current thread.
     *
     * @param action action
     * @param callback callback
     */
    public void schedule(Runnable action, Runnable callback) {
        schedule(() -> { action.run(); return SUCCESS; }, (v) -> { callback.run(); });
    }

    /**
     * Schedules executable action which produces result.
     * Executes "callback" after "action" finished work.
     *
     * Does not block current thread.
     *
     * @param action action
     * @param callback callback
     * @param <R> R
     */
    @SuppressWarnings("unchecked")
    public <R> void schedule(Supplier<R> action, Consumer<R> callback) {
        long expect = cid.incrementAndGet();
        callbacks.put(expect, (Consumer)callback);
        producer.signalEvent(new SupplierEvent(tid, expect, action));
    }

    /**
     * Schedules executable action.
     * Waits for job completion.
     * Returns true if job was executed successfully, returns false otherwise.
     *
     * @param action action
     * @return boolean
     * @throws InterruptedException if terminated
     */
    public boolean await(Runnable action) throws InterruptedException {
        return SUCCESS == await(() -> { action.run(); return SUCCESS; });
    }

    /**
     * Schedules executable action.
     * Waits for job completion and returns generated result.
     *
     * @param action action
     * @param <R> R
     * @return result
     * @throws InterruptedException if terminated
     */
    @SuppressWarnings("unchecked")
    public <R> R await(Supplier<R> action) throws InterruptedException {
        long expect = cid.incrementAndGet();
        EventLock<SupplierResponse> lock = EventLock.singleLock(SupplierResponse.class, consumer);

        producer.signalEvent(new SupplierEvent(tid, expect, action));

        return (R)lock.waitForEvent(e -> e.tid == tid && e.cid == expect).value;
    }

    private static class RunnableEvent implements Event {

        private final long tid;

        private final Runnable action;

        public RunnableEvent(long tid, Runnable action) {
            this.tid = tid;
            this.action = action;
        }
    }

    private static class SupplierEvent implements Event {

        private final long tid;

        private final long cid;

        private final Supplier<?> function;

        public SupplierEvent(long tid, long cid, Supplier<?> function) {
            this.tid = tid;
            this.cid = cid;
            this.function = function;
        }

    }

    private static class SupplierResponse implements Event {

        private final long tid;

        private final long cid;

        private final Object value;

        public SupplierResponse(long tid, long cid, Object value) {
            this.tid = tid;
            this.cid = cid;
            this.value = value;
        }

    }


}
