/*
 *  Licensed to the Apache Software Foundation (ASF) under one
 *  or more contributor license agreements.  See the NOTICE file
 *  distributed with this work for additional information
 *  regarding copyright ownership.  The ASF licenses this file
 *  to you under the Apache License, Version 2.0 (the
 *  "License"); you may not use this file except in compliance
 *  with the License.  You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 *  Unless required by applicable law or agreed to in writing,
 *  software distributed under the License is distributed on an
 *   * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 *  KIND, either express or implied.  See the License for the
 *  specific language governing permissions and limitations
 *  under the License.
 */

package org.apache.synapse.commons.executors;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.axis2.transport.base.threads.NativeThreadFactory;

import java.util.concurrent.*;
import java.util.Map;
import java.util.HashMap;

/**
 * This is the class used for executing the tasks with a given priority. It is backed by a
 * BlockingQueue and a ThreadPoolExecutor. The BlockingQueue is a custom implementation which 
 * has multiple internal queues for handling separate priorities.
 */
public class PriorityExecutor {
    private final Log log = LogFactory.getLog(PriorityExecutor.class);    

    /** Actual thread pool executor */
    private ThreadPoolExecutor executor;
    /** Name of the executor */
    private String name = null;
    /** Core threads count */
    private int core = ExecutorConstants.DEFAULT_CORE;
    /** Max thread count */
    private int max = ExecutorConstants.DEFAULT_MAX;
    /** Keep alive time for spare threads */
    private int keepAlive = ExecutorConstants.DEFAULT_KEEP_ALIVE;
    /** This will be executed before the Task is submitted  */
    private BeforeExecuteHandler beforeExecuteHandler;
    /** Queue used by the executor */
    private MultiPriorityBlockingQueue<Runnable> queue;
    /** this is used by the file based synapse xml configuration */
    private String fileName;
    /** Weather executor is initializer */
    private boolean initialzed;

    /**
     * Execute a given task with the priority specified. If the task throws an exception,
     * it will be captured and logged to prevent the threads from dying. 
     *
     * @param task task to be executed
     * @param priority priority of the task
     */
    public void execute(final Runnable task, int priority) {
        if (!initialzed) {
            throw new IllegalStateException("Executor is not initialized");
        }
        // create a dummy worker to execute the task
        Worker w = new Worker(task, priority);

        if (beforeExecuteHandler != null) {
            beforeExecuteHandler.beforeExecute(w);
        }
        // we are capturing all the exceptions to prevent threads from dying
        executor.execute(w);
    }


    /**
     * Initialize the executor by using the properties. Create the queues
     * and ThreadPool executor.
     */
    public void init() {
        if (queue == null) {
            throw new IllegalStateException("Queue should be specified before initializing");
        }

        executor = new ThreadPoolExecutor(core, max, keepAlive, TimeUnit.SECONDS, queue,
                new NativeThreadFactory(new ThreadGroup("executor-group"),
                        "priority-worker" + (name != null ? "-" + name : "")));

        initialzed = true;

        if (log.isDebugEnabled()) {
            log.debug("Started the thread pool executor with threads, " +
                    "core = " + core + " max = " + max +
                    ", keep-alive = " + keepAlive);
        }
    }

    /**
     * Destroy the executor. Stop all the threads running. 
     */
    public void destroy() {
        if (initialzed) {
            if (log.isDebugEnabled()) {
                log.debug("Shutting down priority executor" + (name != null ? ": " + name : ""));
            }

            executor.shutdown();

            try {
                executor.awaitTermination(100, TimeUnit.MILLISECONDS);
            } catch (InterruptedException e) {
                log.error("Failed to Shut down Executor");
            }

            initialzed = false;
        }
    }

    /**
     * Set the name of the executor
     *
     * @param name of the executor
     */
    public void setName(String name) {
        this.name = name;
    }

    /**
     * Get the name of the executor
     *
     * @return name of the executor
     */
    public String getName() {
        return name;
    }

    /**
     * Set a handler for execute before putting a worker in to the queues.
     * User can set some properties to the worker at this point. This
     * allows users to get more control over the queue selection algorithm.
     * This is an optional configuration.
     *
     * @param beforeExecuteHandler an object implementing the BeforeExecuteHandler 
     */
    public void setBeforeExecuteHandler(
            BeforeExecuteHandler beforeExecuteHandler) {
        this.beforeExecuteHandler = beforeExecuteHandler;
    }

    /**
     * Get the handler that is executed before the worker is put in to the queue
     *
     * @return an object of BeforeExecuteHandler 
     */
    public BeforeExecuteHandler getBeforeExecuteHandler() {
        return beforeExecuteHandler;
    }

    /**
     * Set the queue.
     *
     * @param queue queue used for handling the priorities
     */
    public void setQueue(MultiPriorityBlockingQueue<Runnable> queue) {
        this.queue = queue;
    }

    /**
     * Get the queue.
     *
     * @return queue used for handling multiple priorities
     */
    public MultiPriorityBlockingQueue<Runnable> getQueue() {
        return queue;
    }

    /**
     * Get the core number of threads
     *
     * @return core number of threads
     */
    public int getCore() {
        return core;
    }

    /**
     * Get the max threads
     *
     * @return max thread
     */
    public int getMax() {
        return max;
    }

    /**
     * Get the keep alive time for threads
     *
     * @return keep alive time for threads
     */
    public int getKeepAlive() {
        return keepAlive;
    }

    /**
     * Set the core number of threads
     *
     * @param core core number of threads
     */
    public void setCore(int core) {
        this.core = core;
    }

    /**
     * Set the max number of threads
     *
     * @param max max threads
     */
    public void setMax(int max) {
        this.max = max;
    }

    /**
     * Set the keep alive time for threads
     *
     * @param keepAlive keep alive threads
     */
    public void setKeepAlive(int keepAlive) {
        this.keepAlive = keepAlive;
    }

    /**
     * Get the file used to store this executor config
     *
     * @return file used for storing the config
     */
    public String getFileName() {
        return fileName;
    }

    /**
     * Set the file used to store the config
     *
     * @param fileName file name
     */
    public void setFileName(String fileName) {
        this.fileName = fileName;
    }

    /**
     * Private class for executing the tasks submitted. This class is used for
     * prevent the threads from dying in case of unhandled exceptions. Also
     * this class implements the Importance for carrying the priority.     
     */
    private class Worker implements Runnable, Importance {
        private Runnable runnable = null;

        private Map<String, Object> properties = new HashMap<String, Object>();

        private int priority = 1;

        private Worker(Runnable runnable, int priority) {
            this.priority = priority;
            this.runnable = runnable;
        }

        @Override
        public void run() {
            try {
                runnable.run();
            } catch (Throwable e) {
                log.error("Unhandled exception", e);
            }
        }

        @Override
        public int getPriority() {
            return priority;
        }

        @Override
        public void setPriority(int p) {
            this.priority = p;
        }

        @Override
        public void setProperty(String name, Object value) {
            properties.put(name, value);
        }

        @Override
        public Object getProperty(String name) {
            return properties.get(name);
        }
    }
}
