serverconnection.java

java 开发的邮件服务器平台。支持以下协议。 协议可以修改为自己的专门标识

/***********************************************************************
 * Copyright (c) 2000-2004 The Apache Software Foundation.             *
 * All rights reserved.                                                *
 * ------------------------------------------------------------------- *
 * Licensed 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.james.util.connection;

import java.io.IOException;
import java.io.InterruptedIOException;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

import org.apache.avalon.cornerstone.services.connection.ConnectionHandler;
import org.apache.avalon.cornerstone.services.connection.ConnectionHandlerFactory;
import org.apache.avalon.excalibur.pool.HardResourceLimitingPool;
import org.apache.avalon.excalibur.pool.ObjectFactory;
import org.apache.avalon.excalibur.pool.Pool;
import org.apache.avalon.excalibur.pool.Poolable;
import org.apache.avalon.excalibur.thread.ThreadPool;
import org.apache.avalon.framework.activity.Disposable;
import org.apache.avalon.framework.activity.Initializable;
import org.apache.avalon.framework.component.Component;
import org.apache.avalon.framework.logger.AbstractLogEnabled;
import org.apache.avalon.framework.logger.LogEnabled;


/**
 * Represents a single server socket managed by a connection manager.
 * The connection manager will spawn a single ServerConnection for each
 * server socket that the connection manager is managing.
 *
 */
public class ServerConnection extends AbstractLogEnabled
    implements Component, Initializable, Runnable {

    /**
     * This is a hack to deal with the fact that there appears to be
     * no platform-independent way to break out of a ServerSocket
     * accept() call.  On some platforms closing either the ServerSocket
     * itself, or its associated InputStream, causes the accept
     * method to exit.  Unfortunately, this behavior is not consistent
     * across platforms.  The deal with this, we introduce a polling
     * loop of 20 seconds for the server socket.  This introduces a
     * cost across platforms, but is necessary to maintain cross-platform
     * functionality.
     */
    private static int POLLING_INTERVAL = 20*1000;

    /**
     * The server socket which this connection is managing
     */
    private ServerSocket serverSocket;

    /**
     * The connection handler factory that generates connection
     * handlers to manage client connections to this server socket
     */
    private ConnectionHandlerFactory handlerFactory;

    /**
     * The pool that produces ClientConnectionRunners
     */
    private Pool runnerPool;

    /**
     * The factory used to provide ClientConnectionRunner objects
     */
    private ObjectFactory theRunnerFactory = new ClientConnectionRunnerFactory();

    /**
     * The thread pool used to spawn individual threads used to manage each
     * client connection.
     */
    private ThreadPool connThreadPool;

    /**
     * The timeout for client sockets spawned off this connection.
     */
    private int socketTimeout;

    /**
     * The maximum number of open client connections that this server
     * connection will allow.
     */
    private int maxOpenConn;

    /**
     * A collection of client connection runners.
     */
    private final ArrayList clientConnectionRunners = new ArrayList();

    /**
     * The thread used to manage this server connection.
     */
    private Thread serverConnectionThread;

    /**
     * The sole constructor for a ServerConnection.
     *
     * @param serverSocket the ServerSocket associated with this ServerConnection
     * @param handlerFactory the factory that generates ConnectionHandlers for the client
     *                       connections spawned off this ServerConnection
     * @param threadPool the ThreadPool used to obtain handler threads
     * @param timeout the client idle timeout for this ServerConnection's client connections
     * @param maxOpenConn the maximum number of open client connections allowed for this
     *                    ServerConnection
     */
    public ServerConnection(ServerSocket serverSocket,
                            ConnectionHandlerFactory handlerFactory,
                            ThreadPool threadPool,
                            int timeout,
                            int maxOpenConn) {
        this.serverSocket = serverSocket;
        this.handlerFactory = handlerFactory;
        connThreadPool = threadPool;
        socketTimeout = timeout;
        this.maxOpenConn = maxOpenConn;
    }

    /**
     * @see org.apache.avalon.framework.activity.Initializable#initialize()
     */
    public void initialize() throws Exception {
        runnerPool = new HardResourceLimitingPool(theRunnerFactory, 5, maxOpenConn);
        if (runnerPool instanceof LogEnabled) {
            ((LogEnabled)runnerPool).enableLogging(getLogger());
        }
        ((Initializable)runnerPool).initialize();
    }

    /**
     * The dispose operation is called by the owning ConnectionManager
     * at the end of its lifecycle.  Cleans up the server connection, forcing
     * everything to finish.
     */
    public void dispose() {
        if (getLogger().isDebugEnabled()) {
            getLogger().debug("Disposing server connection..." + this.toString());
        }
        synchronized( this ) {
            if( null != serverConnectionThread ) {
                // Execution of this block means that the run() method
                // hasn't finished yet.  So we interrupt the thread
                // to terminate run() and wait for the run() method
                // to finish.  The notifyAll() at the end of run() will
                // wake this thread and allow dispose() to end.
                Thread thread = serverConnectionThread;
                serverConnectionThread = null;
                thread.interrupt();
                try {
                    serverSocket.close();
                } catch (IOException ie) {
                    // Ignored - we're doing this to break out of the
                    // accept.  This minimizes the time required to
                    // shutdown the server.  Unfortunately, this is
                    // not guaranteed to work on all platforms.  See
                    // the comments for POLLING_INTERVAL
                }
                try {
                    if (POLLING_INTERVAL > 0) {
                        wait(2L*POLLING_INTERVAL);
                    } else {
                        wait();
                    }
                } catch (InterruptedException ie) {
                    // Expected - just complete dispose()
                }
            }
            if (runnerPool instanceof Disposable) {
                ((Disposable)runnerPool).dispose();
            }
            runnerPool = null;
        }

        getLogger().debug("Closed server connection - cleaning up clients - " + this.toString());

        synchronized (clientConnectionRunners) {
            Iterator runnerIterator = clientConnectionRunners.iterator();
            while( runnerIterator.hasNext() ) {
                ClientConnectionRunner runner = (ClientConnectionRunner)runnerIterator.next();
                runner.dispose();
                runner = null;
            }
            clientConnectionRunners.clear();
        }

        getLogger().debug("Cleaned up clients - " + this.toString());

    }

    /**
     * Returns a ClientConnectionRunner in the set managed by this ServerConnection object.
     *
     * @param clientConnectionRunner the ClientConnectionRunner to be added
     */
    private ClientConnectionRunner addClientConnectionRunner()
            throws Exception {
        synchronized (clientConnectionRunners) {
            ClientConnectionRunner clientConnectionRunner = (ClientConnectionRunner)runnerPool.get();
            clientConnectionRunners.add(clientConnectionRunner);
            if (getLogger().isDebugEnabled()) {
                getLogger().debug("Adding one connection for a total of " + clientConnectionRunners.size());
            }
            return clientConnectionRunner;
        }
    }

    /**
     * Removes a ClientConnectionRunner from the set managed by this ServerConnection object.
     *
     * @param clientConnectionRunner the ClientConnectionRunner to be removed
     */
    private void removeClientConnectionRunner(ClientConnectionRunner clientConnectionRunner) {
        synchronized (clientConnectionRunners) {
            if (clientConnectionRunners.remove(clientConnectionRunner)) {
                if (getLogger().isDebugEnabled()) {
                    getLogger().debug("Releasing one connection, leaving a total of " + clientConnectionRunners.size());
                }
                runnerPool.put(clientConnectionRunner);
            }
        }
    }

    /**
     * Provides the body for the thread of execution for a ServerConnection.
     * Connections made to the server socket are passed to an appropriate,
     * newly created, ClientConnectionRunner
     */
    public void run() {
        serverConnectionThread = Thread.currentThread();

        int ioExceptionCount = 0;
        try {
            serverSocket.setSoTimeout(POLLING_INTERVAL);
        } catch (SocketException se) {
            // Ignored - for the moment
        }

        if ((getLogger().isDebugEnabled()) && (serverConnectionThread != null)) {
            StringBuffer debugBuffer =
                new StringBuffer(128)
                    .append(serverConnectionThread.getName())