/*
 *  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.mina.filter.executor;

import java.util.EnumSet;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;

import org.apache.mina.core.filterchain.IoFilterAdapter;
import org.apache.mina.core.filterchain.IoFilterChain;
import org.apache.mina.core.filterchain.IoFilterEvent;
import org.apache.mina.core.session.IdleStatus;
import org.apache.mina.core.session.IoEventType;
import org.apache.mina.core.session.IoSession;
import org.apache.mina.core.write.WriteRequest;

/**
 * A filter that forwards I/O events to {@link Executor} to enforce a certain
 * thread model while allowing the events per session to be processed
 * simultaneously. You can apply various thread model by inserting this filter
 * to a {@link IoFilterChain}.
 * 
 * <h2>Life Cycle Management</h2>
 * 
 * Please note that this filter doesn't manage the life cycle of the {@link Executor}.
 * If you created this filter using {@link #ExecutorFilter(Executor)} or similar
 * constructor that accepts an {@link Executor} that you've instantiated, you have
 * full control and responsibility of managing its life cycle (e.g. calling
 * {@link ExecutorService#shutdown()}.
 * <p> 
 * If you created this filter using convenience constructors like
 * {@link #ExecutorFilter(int)}, then you can shut down the executor by calling
 * {@link #destroy()} explicitly.
 * 
 * <h2>Event Ordering</h2>
 * 
 * All convenience constructors of this filter creates a new
 * {@link OrderedThreadPoolExecutor} instance.  Therefore, the order of event is
 * maintained like the following:
 * <ul>
 * <li>All event handler methods are called exclusively.
 *     (e.g. messageReceived and messageSent can't be invoked at the same time.)</li>
 * <li>The event order is never mixed up.
 *     (e.g. messageReceived is always invoked before sessionClosed or messageSent.)</li>
 * </ul>
 * However, if you specified other {@link Executor} instance in the constructor,
 * the order of events are not maintained at all.  This means more than one event
 * handler methods can be invoked at the same time with mixed order.  For example,
 * let's assume that messageReceived, messageSent, and sessionClosed events are
 * fired.
 * <ul>
 * <li>All event handler methods can be called simultaneously.
 *     (e.g. messageReceived and messageSent can be invoked at the same time.)</li>
 * <li>The event order can be mixed up.
 *     (e.g. sessionClosed or messageSent can be invoked before messageReceived
 *           is invoked.)</li>
 * </ul>
 * If you need to maintain the order of events per session, please specify an
 * {@link OrderedThreadPoolExecutor} instance or use the convenience constructors.
 * 
 * <h2>Selective Filtering</h2>
 * 
 * By default, all event types but <tt>sessionCreated</tt>, <tt>filterWrite</tt>,
 * <tt>filterClose</tt> and <tt>filterSetTrafficMask</tt> are submitted to the
 * underlying executor, which is most common setting.
 * <p>
 * If you want to submit only a certain set of event types, you can specify them
 * in the constructor.  For example, you could configure a thread pool for
 * write operation for the maximum performance:
 * <pre><code>
 * IoService service = ...;
 * DefaultIoFilterChainBuilder chain = service.getFilterChain();
 * 
 * chain.addLast("codec", new ProtocolCodecFilter(...));
 * // Use one thread pool for most events.
 * chain.addLast("executor1", new ExecutorFilter());
 * // and another dedicated thread pool for 'filterWrite' events.
 * chain.addLast("executor2", new ExecutorFilter(IoEventType.WRITE));
 * </code></pre>
 * 
 * <h2>Preventing {@link OutOfMemoryError}</h2>
 * 
 * Please refer to {@link IoEventQueueThrottle}, which is specified as
 * a parameter of the convenience constructors.
 * 
 * @author <a href="http://mina.apache.org">Apache MINA Project</a>
 * 
 * @see OrderedThreadPoolExecutor
 * @see UnorderedThreadPoolExecutor
 * @org.apache.xbean.XBean
 */
public class ExecutorFilter extends IoFilterAdapter {
    /** The list of handled events */
    private EnumSet<IoEventType> eventTypes;

    /** The associated executor */
    private Executor executor;

    /** A flag set if the executor can be managed */
    private boolean manageableExecutor;

    /** The default pool size */
    private static final int DEFAULT_MAX_POOL_SIZE = 16;

    /** The number of thread to create at startup */
    private static final int BASE_THREAD_NUMBER = 0;

    /** The default KeepAlive time, in seconds */
    private static final long DEFAULT_KEEPALIVE_TIME = 30;

    /** 
     * A set of flags used to tell if the Executor has been created 
     * in the constructor or passed as an argument. In the second case, 
     * the executor state can be managed.
     **/
    private static final boolean MANAGEABLE_EXECUTOR = true;

    private static final boolean NOT_MANAGEABLE_EXECUTOR = false;

    /** A list of default EventTypes to be handled by the executor */
    private static final IoEventType[] DEFAULT_EVENT_SET = new IoEventType[] { IoEventType.EXCEPTION_CAUGHT,
            IoEventType.MESSAGE_RECEIVED, IoEventType.MESSAGE_SENT, IoEventType.SESSION_CLOSED,
            IoEventType.SESSION_IDLE, IoEventType.SESSION_OPENED };

    /**
     * (Convenience constructor) Creates a new instance with a new
     * {@link OrderedThreadPoolExecutor}, no thread in the pool, and a 
     * maximum of 16 threads in the pool. All the event will be handled 
     * by this default executor.
     */
    public ExecutorFilter() {
        // Create a new default Executor
        Executor executor = createDefaultExecutor(BASE_THREAD_NUMBER, DEFAULT_MAX_POOL_SIZE, DEFAULT_KEEPALIVE_TIME,
                TimeUnit.SECONDS, Executors.defaultThreadFactory(), null);

        // Initialize the filter
        init(executor, MANAGEABLE_EXECUTOR);
    }

    /**
     * (Convenience constructor) Creates a new instance with a new
     * {@link OrderedThreadPoolExecutor}, no thread in the pool, but 
     * a maximum of threads in the pool is given. All the event will be handled 
     * by this default executor.
     * 
     * @param maximumPoolSize The maximum pool size
     */
    public ExecutorFilter(int maximumPoolSize) {
        // Create a new default Executor
        Executor executor = createDefaultExecutor(BASE_THREAD_NUMBER, maximumPoolSize, DEFAULT_KEEPALIVE_TIME,
                TimeUnit.SECONDS, Executors.defaultThreadFactory(), null);

        // Initialize the filter
        init(executor, MANAGEABLE_EXECUTOR);
    }

    /**
     * (Convenience constructor) Creates a new instance with a new
     * {@link OrderedThreadPoolExecutor}, a number of thread to start with, a  
     * maximum of threads the pool can contain. All the event will be handled 
     * by this default executor.
     *
     * @param corePoolSize The initial pool size
     * @param maximumPoolSize The maximum pool size
     */
    public ExecutorFilter(int corePoolSize, int maximumPoolSize) {
        // Create a new default Executor
        Executor executor = createDefaultExecutor(corePoolSize, maximumPoolSize, DEFAULT_KEEPALIVE_TIME,
                TimeUnit.SECONDS, Executors.defaultThreadFactory(), null);

        // Initialize the filter
        init(executor, MANAGEABLE_EXECUTOR);
    }

    /**
     * (Convenience constructor) Creates a new instance with a new
     * {@link OrderedThreadPoolExecutor}.
     * 
     * @param corePoolSize The initial pool size
     * @param maximumPoolSize The maximum pool size
     * @param keepAliveTime Default duration for a thread
     * @param unit Time unit used for the keepAlive value
     */
    public ExecutorFilter(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit) {
        // Create a new default Executor
        Executor executor = createDefaultExecutor(corePoolSize, maximumPoolSize, keepAliveTime, unit,
                Executors.defaultThreadFactory(), null);

        // Initialize the filter
        init(executor, MANAGEABLE_EXECUTOR);
    }

    /**
     * (Convenience constructor) Creates a new instance with a new
     * {@link OrderedThreadPoolExecutor}.
     * 
     * @param corePoolSize The initial pool size
     * @param maximumPoolSize The maximum pool size
     * @param keepAliveTime Default duration for a thread
     * @param unit Time unit used for the keepAlive value
     * @param queueHandler The queue used to store events
     */
    public ExecutorFilter(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit,
            IoEventQueueHandler queueHandler) {
        // Create a new default Executor
        Executor executor = createDefaultExecutor(corePoolSize, maximumPoolSize, keepAliveTime, unit,
                Executors.defaultThreadFactory(), queueHandler);

        // Initialize the filter
        init(executor, MANAGEABLE_EXECUTOR);
    }

    /**
     * (Convenience constructor) Creates a new instance with a new
     * {@link OrderedThreadPoolExecutor}.
     * 
     * @param corePoolSize The initial pool size
     * @param maximumPoolSize The maximum pool size
     * @param keepAliveTime Default duration for a thread
     * @param unit Time unit used for the keepAlive value
     * @param threadFactory The factory used to create threads
     */
    public ExecutorFilter(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit,
            ThreadFactory threadFactory) {
        // Create a new default Executor
        Executor executor = createDefaultExecutor(corePoolSize, maximumPoolSize, keepAliveTime, unit, threadFactory,
                null);

        // Initialize the filter
        init(executor, MANAGEABLE_EXECUTOR);
    }

    /**
     * (Convenience constructor) Creates a new instance with a new
     * {@link OrderedThreadPoolExecutor}.
     * 
     * @param corePoolSize The initial pool size
     * @param maximumPoolSize The maximum pool size
     * @param keepAliveTime Default duration for a thread
     * @param unit Time unit used for the keepAlive value
     * @param threadFactory The factory used to create threads
     * @param queueHandler The queue used to store events
     */
    public ExecutorFilter(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit,
            ThreadFactory threadFactory, IoEventQueueHandler queueHandler) {
        // Create a new default Executor
        Executor executor = new OrderedThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, unit,
                threadFactory, queueHandler);

        // Initialize the filter
        init(executor, MANAGEABLE_EXECUTOR);
    }

    /**
     * (Convenience constructor) Creates a new instance with a new
     * {@link OrderedThreadPoolExecutor}.
     * 
     * @param eventTypes The event for which the executor will be used
     */
    public ExecutorFilter(IoEventType... eventTypes) {
        // Create a new default Executor
        Executor executor = createDefaultExecutor(BASE_THREAD_NUMBER, DEFAULT_MAX_POOL_SIZE, DEFAULT_KEEPALIVE_TIME,
                TimeUnit.SECONDS, Executors.defaultThreadFactory(), null);

        // Initialize the filter
        init(executor, MANAGEABLE_EXECUTOR, eventTypes);
    }

    /**
     * (Convenience constructor) Creates a new instance with a new
     * {@link OrderedThreadPoolExecutor}.
     * 
     * @param maximumPoolSize The maximum pool size
     * @param eventTypes The event for which the executor will be used
     */
    public ExecutorFilter(int maximumPoolSize, IoEventType... eventTypes) {
        // Create a new default Executor
        Executor executor = createDefaultExecutor(BASE_THREAD_NUMBER, maximumPoolSize, DEFAULT_KEEPALIVE_TIME,
                TimeUnit.SECONDS, Executors.defaultThreadFactory(), null);

        // Initialize the filter
        init(executor, MANAGEABLE_EXECUTOR, eventTypes);
    }

    /**
     * (Convenience constructor) Creates a new instance with a new
     * {@link OrderedThreadPoolExecutor}.
     * 
     * @param corePoolSize The initial pool size
     * @param maximumPoolSize The maximum pool size
     * @param eventTypes The event for which the executor will be used
     */
    public ExecutorFilter(int corePoolSize, int maximumPoolSize, IoEventType... eventTypes) {
        // Create a new default Executor
        Executor executor = createDefaultExecutor(corePoolSize, maximumPoolSize, DEFAULT_KEEPALIVE_TIME,
                TimeUnit.SECONDS, Executors.defaultThreadFactory(), null);

        // Initialize the filter
        init(executor, MANAGEABLE_EXECUTOR, eventTypes);
    }

    /**
     * (Convenience constructor) Creates a new instance with a new
     * {@link OrderedThreadPoolExecutor}.
     * 
     * @param corePoolSize The initial pool size
     * @param maximumPoolSize The maximum pool size
     * @param keepAliveTime Default duration for a thread
     * @param unit Time unit used for the keepAlive value
     * @param eventTypes The event for which the executor will be used
     */
    public ExecutorFilter(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit,
            IoEventType... eventTypes) {
        // Create a new default Executor
        Executor executor = createDefaultExecutor(corePoolSize, maximumPoolSize, keepAliveTime, unit,
                Executors.defaultThreadFactory(), null);

        // Initialize the filter
        init(executor, MANAGEABLE_EXECUTOR, eventTypes);
    }

    /**
     * (Convenience constructor) Creates a new instance with a new
     * {@link OrderedThreadPoolExecutor}.
     * 
     * @param corePoolSize The initial pool size
     * @param maximumPoolSize The maximum pool size
     * @param keepAliveTime Default duration for a thread
     * @param unit Time unit used for the keepAlive value
     * @param queueHandler The queue used to store events
     * @param eventTypes The event for which the executor will be used
     */
    public ExecutorFilter(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit,
            IoEventQueueHandler queueHandler, IoEventType... eventTypes) {
        // Create a new default Executor
        Executor executor = createDefaultExecutor(corePoolSize, maximumPoolSize, keepAliveTime, unit,
                Executors.defaultThreadFactory(), queueHandler);

        // Initialize the filter
        init(executor, MANAGEABLE_EXECUTOR, eventTypes);
    }

    /**
     * (Convenience constructor) Creates a new instance with a new
     * {@link OrderedThreadPoolExecutor}.
     * 
     * @param corePoolSize The initial pool size
     * @param maximumPoolSize The maximum pool size
     * @param keepAliveTime Default duration for a thread
     * @param unit Time unit used for the keepAlive value
     * @param threadFactory The factory used to create threads
     * @param eventTypes The event for which the executor will be used
     */
    public ExecutorFilter(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit,
            ThreadFactory threadFactory, IoEventType... eventTypes) {
        // Create a new default Executor
        Executor executor = createDefaultExecutor(corePoolSize, maximumPoolSize, keepAliveTime, unit, threadFactory,
                null);

        // Initialize the filter
        init(executor, MANAGEABLE_EXECUTOR, eventTypes);
    }

    /**
     * (Convenience constructor) Creates a new instance with a new
     * {@link OrderedThreadPoolExecutor}.
     * 
     * @param corePoolSize The initial pool size
     * @param maximumPoolSize The maximum pool size
     * @param keepAliveTime Default duration for a thread
     * @param unit Time unit used for the keepAlive value
     * @param threadFactory The factory used to create threads
     * @param queueHandler The queue used to store events
     * @param eventTypes The event for which the executor will be used
     */
    public ExecutorFilter(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit,
            ThreadFactory threadFactory, IoEventQueueHandler queueHandler, IoEventType... eventTypes) {
        // Create a new default Executor
        Executor executor = new OrderedThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, unit,
                threadFactory, queueHandler);

        // Initialize the filter
        init(executor, MANAGEABLE_EXECUTOR, eventTypes);
    }

    /**
     * Creates a new instance with the specified {@link Executor}.
     * 
     * @param executor the user's managed Executor to use in this filter
     */
    public ExecutorFilter(Executor executor) {
        // Initialize the filter
        init(executor, NOT_MANAGEABLE_EXECUTOR);
    }

    /**
     * Creates a new instance with the specified {@link Executor}.
     * 
     * @param executor the user's managed Executor to use in this filter
     * @param eventTypes The event for which the executor will be used
     */
    public ExecutorFilter(Executor executor, IoEventType... eventTypes) {
        // Initialize the filter
        init(executor, NOT_MANAGEABLE_EXECUTOR, eventTypes);
    }

    /**
     * Create an OrderedThreadPool executor.
     *
     * @param corePoolSize The initial pool sizePoolSize
     * @param maximumPoolSize The maximum pool size
     * @param keepAliveTime Default duration for a thread
     * @param unit Time unit used for the keepAlive value
     * @param threadFactory The factory used to create threads
     * @param queueHandler The queue used to store events
     * @return An instance of the created Executor
     */
    private Executor createDefaultExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit,
            ThreadFactory threadFactory, IoEventQueueHandler queueHandler) {
        // Create a new Executor
        Executor executor = new OrderedThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, unit,
                threadFactory, queueHandler);

        return executor;
    }

    /**
     * Create an EnumSet from an array of EventTypes, and set the associated
     * eventTypes field.
     *
     * @param eventTypes The array of handled events
     */
    private void initEventTypes(IoEventType... eventTypes) {
        if ((eventTypes == null) || (eventTypes.length == 0)) {
            eventTypes = DEFAULT_EVENT_SET;
        }

        // Copy the list of handled events in the event set
        this.eventTypes = EnumSet.of(eventTypes[0], eventTypes);

        // Check that we don't have the SESSION_CREATED event in the set
        if (this.eventTypes.contains(IoEventType.SESSION_CREATED)) {
            this.eventTypes = null;
            throw new IllegalArgumentException(IoEventType.SESSION_CREATED + " is not allowed.");
        }
    }

    /**
     * Creates a new instance of ExecutorFilter. This private constructor is called by all
     * the public constructor.
     *
     * @param executor The underlying {@link Executor} in charge of managing the Thread pool.
     * @param manageableExecutor Tells if the Executor's Life Cycle can be managed or not
     * @param eventTypes The lit of event which are handled by the executor
     */
    private void init(Executor executor, boolean manageableExecutor, IoEventType... eventTypes) {
        if (executor == null) {
            throw new IllegalArgumentException("executor");
        }

        initEventTypes(eventTypes);
        this.executor = executor;
        this.manageableExecutor = manageableExecutor;
    }

    /**
     * Shuts down the underlying executor if this filter hase been created via
     * a convenience constructor.
     */
    @Override
    public void destroy() {
        if (manageableExecutor) {
            ((ExecutorService) executor).shutdown();
        }
    }

    /**
     * @return the underlying {@link Executor} instance this filter uses.
     */
    public final Executor getExecutor() {
        return executor;
    }

    /**
     * Fires the specified event through the underlying executor.
     * 
     * @param event The filtered event
     */
    protected void fireEvent(IoFilterEvent event) {
        executor.execute(event);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void onPreAdd(IoFilterChain parent, String name, NextFilter nextFilter) throws Exception {
        if (parent.contains(this)) {
            throw new IllegalArgumentException(
                    "You can't add the same filter instance more than once.  Create another instance and add it.");
        }
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public final void sessionOpened(NextFilter nextFilter, IoSession session) {
        if (eventTypes.contains(IoEventType.SESSION_OPENED)) {
            IoFilterEvent event = new IoFilterEvent(nextFilter, IoEventType.SESSION_OPENED, session, null);
            fireEvent(event);
        } else {
            nextFilter.sessionOpened(session);
        }
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public final void sessionClosed(NextFilter nextFilter, IoSession session) {
        if (eventTypes.contains(IoEventType.SESSION_CLOSED)) {
            IoFilterEvent event = new IoFilterEvent(nextFilter, IoEventType.SESSION_CLOSED, session, null);
            fireEvent(event);
        } else {
            nextFilter.sessionClosed(session);
        }
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public final void sessionIdle(NextFilter nextFilter, IoSession session, IdleStatus status) {
        if (eventTypes.contains(IoEventType.SESSION_IDLE)) {
            IoFilterEvent event = new IoFilterEvent(nextFilter, IoEventType.SESSION_IDLE, session, status);
            fireEvent(event);
        } else {
            nextFilter.sessionIdle(session, status);
        }
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public final void exceptionCaught(NextFilter nextFilter, IoSession session, Throwable cause) {
        if (eventTypes.contains(IoEventType.EXCEPTION_CAUGHT)) {
            IoFilterEvent event = new IoFilterEvent(nextFilter, IoEventType.EXCEPTION_CAUGHT, session, cause);
            fireEvent(event);
        } else {
            nextFilter.exceptionCaught(session, cause);
        }
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public final void messageReceived(NextFilter nextFilter, IoSession session, Object message) {
        if (eventTypes.contains(IoEventType.MESSAGE_RECEIVED)) {
            IoFilterEvent event = new IoFilterEvent(nextFilter, IoEventType.MESSAGE_RECEIVED, session, message);
            fireEvent(event);
        } else {
            nextFilter.messageReceived(session, message);
        }
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public final void messageSent(NextFilter nextFilter, IoSession session, WriteRequest writeRequest) {
        if (eventTypes.contains(IoEventType.MESSAGE_SENT)) {
            IoFilterEvent event = new IoFilterEvent(nextFilter, IoEventType.MESSAGE_SENT, session, writeRequest);
            fireEvent(event);
        } else {
            nextFilter.messageSent(session, writeRequest);
        }
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public final void filterWrite(NextFilter nextFilter, IoSession session, WriteRequest writeRequest) {
        if (eventTypes.contains(IoEventType.WRITE)) {
            IoFilterEvent event = new IoFilterEvent(nextFilter, IoEventType.WRITE, session, writeRequest);
            fireEvent(event);
        } else {
            nextFilter.filterWrite(session, writeRequest);
        }
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public final void filterClose(NextFilter nextFilter, IoSession session) throws Exception {
        if (eventTypes.contains(IoEventType.CLOSE)) {
            IoFilterEvent event = new IoFilterEvent(nextFilter, IoEventType.CLOSE, session, null);
            fireEvent(event);
        } else {
            nextFilter.filterClose(session);
        }
    }
}