executorfilter.java

mina是以Java实现的一个开源的网络程序框架

            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, DEFAULT_EVENT_SET);
    }

    /**
     * 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
     * @param
     */
    private void init(Executor executor, boolean manageableExecutor, IoEventType... eventTypes) {
        if (executor == null) {
            throw new NullPointerException("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();
        }
    }

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

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

    /**
     * A trigger fired when adding this filter in a chain. As this filter can be
     * added only once in a chain, if the chain already contains the same filter,
     * and exception will be thrown.
     * 
     * @param parent The chain in which we want to inject this filter
     * @param name The Fitler's name
     * @param nextFilter The next filter in the chain
     * 
     * @throws IllegalArgumentException If the filter is already present in the chain
     */
    @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);
        }
    }
}