package org.mockserver.scheduler;

import com.google.common.annotations.VisibleForTesting;
import org.mockserver.configuration.Configuration;
import org.mockserver.httpclient.SocketCommunicationException;
import org.mockserver.log.model.LogEntry;
import org.mockserver.logging.MockServerLogger;
import org.mockserver.mock.action.http.HttpForwardActionResult;
import org.mockserver.model.BinaryMessage;
import org.mockserver.model.Delay;
import org.mockserver.model.HttpResponse;
import org.slf4j.event.Level;

import java.util.concurrent.*;
import java.util.function.BiConsumer;
import java.util.function.Predicate;

import static java.util.concurrent.TimeUnit.MILLISECONDS;
import static org.mockserver.log.model.LogEntry.LogMessageType.WARN;
import static org.mockserver.mock.HttpState.getPort;
import static org.mockserver.mock.HttpState.setPort;

/**
 * @author jamesdbloom
 */
public class Scheduler {

    private final Configuration configuration;
    private final ScheduledExecutorService scheduler;

    private final boolean synchronous;

    public static class SchedulerThreadFactory implements ThreadFactory {

        private final String name;
        private final boolean daemon;
        private static int threadInitNumber;

        public SchedulerThreadFactory(String name) {
            this.name = name;
            this.daemon = true;
        }

        public SchedulerThreadFactory(String name, boolean daemon) {
            this.name = name;
            this.daemon = daemon;
        }

        @Override
        @SuppressWarnings("NullableProblems")
        public Thread newThread(Runnable runnable) {
            Thread thread = new Thread(runnable, "MockServer-" + name + threadInitNumber++);
            thread.setDaemon(daemon);
            return thread;
        }
    }

    private final MockServerLogger mockServerLogger;

    public Scheduler(Configuration configuration, MockServerLogger mockServerLogger) {
        this(configuration, mockServerLogger, false);
    }

    @VisibleForTesting
    public Scheduler(Configuration configuration, MockServerLogger mockServerLogger, boolean synchronous) {
        this.configuration = configuration;
        this.mockServerLogger = mockServerLogger;
        this.synchronous = synchronous;
        if (!this.synchronous) {
            this.scheduler = new ScheduledThreadPoolExecutor(
                configuration.actionHandlerThreadCount(),
                new SchedulerThreadFactory("Scheduler"),
                new ThreadPoolExecutor.CallerRunsPolicy()
            );
        } else {
            this.scheduler = null;
        }
    }

    public synchronized void shutdown() {
        if (!scheduler.isShutdown()) {
            scheduler.shutdown();
            try {
                scheduler.awaitTermination(500, MILLISECONDS);
            } catch (InterruptedException ignore) {
                // ignore interrupted exception
            }
        }
    }

    private void run(Runnable command, Integer port) {
        setPort(port);
        try {
            command.run();
        } catch (Throwable throwable) {
            if (mockServerLogger.isEnabledForInstance(Level.INFO)) {
                mockServerLogger.logEvent(
                    new LogEntry()
                        .setType(WARN)
                        .setLogLevel(Level.INFO)
                        .setMessageFormat(throwable.getMessage())
                        .setThrowable(throwable)
                );
            }
        }
    }

    public void submitAsync(Runnable command, Delay... delays) {
        long delayMillis = sampleCombinedDelayMillis(delays);
        Integer port = getPort();
        if (scheduler != null) {
            if (delayMillis > 0) {
                scheduler.schedule(() -> run(command, port), delayMillis, MILLISECONDS);
            } else {
                scheduler.submit(() -> run(command, port));
            }
        } else {
            if (delayMillis > 0) {
                try {
                    MILLISECONDS.sleep(delayMillis);
                } catch (InterruptedException ie) {
                    throw new RuntimeException("InterruptedException while applying delay", ie);
                }
            }
            run(command, port);
        }
    }

    public void schedule(Runnable command, boolean synchronous, Delay... delays) {
        long delayMillis = sampleCombinedDelayMillis(delays);
        Integer port = getPort();
        if (this.synchronous || synchronous) {
            if (delayMillis > 0) {
                try {
                    MILLISECONDS.sleep(delayMillis);
                } catch (InterruptedException ie) {
                    throw new RuntimeException("InterruptedException while apply delay to response", ie);
                }
            }
            run(command, port);
        } else {
            if (delayMillis > 0) {
                scheduler.schedule(() -> run(command, port), delayMillis, MILLISECONDS);
            } else {
                run(command, port);
            }
        }
    }

    private long sampleCombinedDelayMillis(Delay... delays) {
        if (delays == null || delays.length == 0) {
            return 0;
        } else if (delays.length == 1 && delays[0] != null) {
            return delays[0].sampleValueMillis();
        } else if (delays.length == 2 && delays[0] == delays[1]) {
            return delays[0] != null ? delays[0].sampleValueMillis() : 0;
        } else {
            long timeInMilliseconds = 0;
            for (Delay delay : delays) {
                if (delay != null) {
                    timeInMilliseconds = Math.min(Long.MAX_VALUE, timeInMilliseconds + delay.sampleValueMillis());
                }
            }
            return timeInMilliseconds;
        }
    }

    public void submit(Runnable command) {
        submit(command, false);
    }

    public void submit(Runnable command, boolean synchronous) {
        Integer port = getPort();
        if (this.synchronous || synchronous) {
            run(command, port);
        } else {
            scheduler.submit(() -> run(command, port));
        }
    }

    public void submit(HttpForwardActionResult future, Runnable command, boolean synchronous, Predicate<Throwable> logException) {
        Integer port = getPort();
        if (future != null) {
            if (this.synchronous || synchronous) {
                try {
                    future.getHttpResponse().get(configuration.maxSocketTimeoutInMillis(), MILLISECONDS);
                } catch (TimeoutException e) {
                    future.getHttpResponse().completeExceptionally(new SocketCommunicationException("Response was not received after " + configuration.maxSocketTimeoutInMillis() + " milliseconds, to make the proxy wait longer please use \"mockserver.maxSocketTimeout\" system property or configuration.maxSocketTimeout(long milliseconds)", e.getCause()));
                } catch (InterruptedException | ExecutionException ex) {
                    future.getHttpResponse().completeExceptionally(ex);
                }
                run(command, port);
            } else {
                future.getHttpResponse().whenCompleteAsync((httpResponse, throwable) -> {
                    if (throwable != null && mockServerLogger.isEnabledForInstance(Level.INFO) && logException.test(throwable)) {
                        mockServerLogger.logEvent(
                            new LogEntry()
                                .setType(WARN)
                                .setLogLevel(Level.INFO)
                                .setMessageFormat(throwable.getMessage())
                                .setThrowable(throwable)
                        );
                    }
                    run(command, port);
                }, scheduler);
            }
        }
    }

    public void submit(CompletableFuture<BinaryMessage> future, Runnable command, boolean synchronous) {
        Integer port = getPort();
        if (future != null) {
            if (this.synchronous || synchronous) {
                try {
                    future.get(configuration.maxSocketTimeoutInMillis(), MILLISECONDS);
                } catch (TimeoutException e) {
                    future.completeExceptionally(new SocketCommunicationException("Response was not received after " + configuration.maxSocketTimeoutInMillis() + " milliseconds, to make the proxy wait longer please use \"mockserver.maxSocketTimeout\" system property or ConfigurationProperties.maxSocketTimeout(long milliseconds)", e.getCause()));
                } catch (InterruptedException | ExecutionException ex) {
                    future.completeExceptionally(ex);
                }
                run(command, port);
            } else {
                future.whenCompleteAsync((httpResponse, throwable) -> command.run(), scheduler);
            }
        }
    }

    public void submit(HttpForwardActionResult future, BiConsumer<HttpResponse, Throwable> consumer, boolean synchronous) {
        if (future != null) {
            if (this.synchronous || synchronous) {
                HttpResponse httpResponse = null;
                Throwable exception = null;
                try {
                    httpResponse = future.getHttpResponse().get(configuration.maxSocketTimeoutInMillis(), MILLISECONDS);
                } catch (TimeoutException e) {
                    exception = new SocketCommunicationException("Response was not received after " + configuration.maxSocketTimeoutInMillis() + " milliseconds, to make the proxy wait longer please use \"mockserver.maxSocketTimeout\" system property or ConfigurationProperties.maxSocketTimeout(long milliseconds)", e.getCause());
                } catch (InterruptedException | ExecutionException ex) {
                    exception = ex;
                }
                try {
                    consumer.accept(httpResponse, exception);
                } catch (Throwable throwable) {
                    if (mockServerLogger.isEnabledForInstance(Level.INFO)) {
                        mockServerLogger.logEvent(
                            new LogEntry()
                                .setType(WARN)
                                .setLogLevel(Level.INFO)
                                .setMessageFormat(throwable.getMessage())
                                .setThrowable(throwable)
                        );
                    }
                }
            } else {
                future.getHttpResponse().whenCompleteAsync(consumer, scheduler);
            }
        }
    }

}
