/*****************************************************************
    * 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.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 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.excalibur.thread.ThreadPool ;
  import org.apache.avalon.framework.activity.Initializable;
  import org.apache.avalon.framework.container.ContainerUtil;
  import org.apache.avalon.framework.logger.AbstractLogEnabled;
  
  
  /***
   * 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 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);
         ContainerUtil.enableLogging(runnerPool,getLogger());
         ContainerUtil.initialize(runnerPool);
     }
 
     /***
      * 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()
                 }
             }
             ContainerUtil.dispose(runnerPool);
             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) {
 
        /*
         * checking runnerPool avoids 'dead-lock' when service is disposing :
         * (dispose() calls dispose all runners)
         * but runner is 'running' and cleans up on exit
         * this situation will result in a dead-lock on 'clientConnectionRunners'
         */
         if( runnerPool == null ) {
             getLogger().info("ServerConnection.removeClientConnectionRunner - dispose has been called - so just return : " + clientConnectionRunner );
             return;
         }
         
         synchronized (clientConnectionRunners) {
             if (clientConnectionRunners.remove(clientConnectionRunner)) {
                 if (getLogger().isDebugEnabled()) {
                     getLogger().debug("Releasing one connection, leaving a total of " + clientConnectionRunners.size());
                 }
                 runnerPool.put(clientConnectionRunner);
             }
         }
 
         synchronized (this) { notify(); } // match the wait(...) in the run() inner loop before accept().
     }
 
     /***
      * 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())
                     .append(" is listening on ")
                     .append(serverSocket.toString());
             getLogger().debug(debugBuffer.toString());
         }
         while( !Thread.currentThread().interrupted() && null != serverConnectionThread ) {
             try {
                 Socket clientSocket = null;
                 try {
                     while (maxOpenConn > 0 && clientConnectionRunners.size() >= maxOpenConn) {
                         getLogger().warn("Maximum number of open connections (" +  clientConnectionRunners.size() + ") in use.");
                         synchronized (this) { wait(10000); }
                     }
 
                     clientSocket = serverSocket.accept();
 
                 } catch( InterruptedIOException iioe ) {
                     // This exception is expected upon ServerConnection shutdown.
                     // See the POLLING_INTERVAL comment
                     continue;
                 } catch( IOException se ) {
                     if (ioExceptionCount > 0) {
                         getLogger().error( "Fatal exception while listening on server socket.  Terminating connection.", se );
                         break;
                     } else {
                         continue;
                     }
                 } catch( SecurityException se ) {
                     getLogger().error( "Fatal exception while listening on server socket.  Terminating connection.", se );
                     break;
                 }
                 ClientConnectionRunner runner = null;
                 synchronized (clientConnectionRunners) {
                     if ((maxOpenConn > 0) && (clientConnectionRunners.size() >= maxOpenConn)) {
                         if (getLogger().isWarnEnabled()) {
                            getLogger().warn("Maximum number of open connections exceeded - refusing connection.  Current number of connections is " + clientConnectionRunners.size());
                            if (getLogger().isWarnEnabled()) {
                                Iterator runnerIterator = clientConnectionRunners.iterator();
                                getLogger().info("Connections: ");
                                while( runnerIterator.hasNext() ) {
                                    getLogger().info("    " + ((ClientConnectionRunner)runnerIterator.next()).toString());
                                }
                            }
                         }
                         try {
                             clientSocket.close();
                         } catch (IOException ignored) {
                             // We ignore this exception, as we already have an error condition.
                         }
                         continue;
                     } else {
                         clientSocket.setSoTimeout(socketTimeout);
                         runner = addClientConnectionRunner();
                         runner.setSocket(clientSocket);
                     }
                 }
                 setupLogger( runner );
                 try {
                     connThreadPool.execute( runner );
                 } catch (Exception e) {
                     // This error indicates that the underlying thread pool
                     // is out of threads.  For robustness, we catch this and
                     // cleanup
                     getLogger().error("Internal error - insufficient threads available to service request.  " +
                                       Thread.activeCount() + " threads in service request pool.", e);
                     try {
                         clientSocket.close();
                     } catch (IOException ignored) {
                         // We ignore this exception, as we already have an error condition.
                     }
                     // In this case, the thread will not remove the client connection runner,
                     // so we must.
                     removeClientConnectionRunner(runner);
                 }
             } catch( IOException ioe ) {
                 getLogger().error( "Exception accepting connection", ioe );
             } catch( Throwable e ) {
                 getLogger().error( "Exception executing client connection runner: " + e.getMessage(), e );
             }
         }
         synchronized( this ) {
             serverConnectionThread = null;
             Thread.currentThread().interrupted();
             notifyAll();
         }
     }
 
     /***
      * An inner class to provide the actual body of the thread of execution
      * that occurs upon a client connection.
      *
      */
     class ClientConnectionRunner extends AbstractLogEnabled
         implements Poolable, Runnable  {
 
         /***
          * The Socket that this client connection is using for transport.
          */
         private Socket clientSocket;
 
         /***
          * The thread of execution associated with this client connection.
          */
         private Thread clientSocketThread;
 
         /***
          * Returns string for diagnostic logging
          */
         public String toString() {
             return getClass().getName() + " for " + clientSocket + " on " + clientSocketThread;
         }
 
         public ClientConnectionRunner() {
         }
 
         /***
          * The dispose operation that terminates the runner.  Should only be
          * called by the ServerConnection that owns the ClientConnectionRunner
          */
         public void dispose() {
             synchronized( this ) {
                 if (null != clientSocketThread) {
                     // 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.
                     clientSocketThread.interrupt();
                     clientSocketThread = null;
                     try {
                         wait();
                     } catch (InterruptedException ie) {
                         // Expected - return from the method
                     }
                 }
             }
         }
 
         /***
          * Sets the socket for a ClientConnectionRunner.
          *
          * @param socket the client socket associated with this ClientConnectionRunner
          */
         public void setSocket(Socket socket) {
             clientSocket = socket;
         }
 
         /***
          * Provides the body for the thread of execution dealing with a particular client
          * connection.  An appropriate ConnectionHandler is created, applied, executed,
          * and released.
          */
         public void run() {
             ConnectionHandler handler = null;
             try {
                 clientSocketThread = Thread.currentThread();
 
                 handler = ServerConnection.this.handlerFactory.createConnectionHandler();
                 String connectionString = null;
                 if( getLogger().isDebugEnabled() ) {
                     connectionString = getConnectionString();
                     String message = "Starting " + connectionString;
                     getLogger().debug( message );
                 }
 
                 handler.handleConnection(clientSocket);
 
                 if( getLogger().isDebugEnabled() ) {
                     String message = "Ending " + connectionString;
                     getLogger().debug( message );
                 }
 
             } catch( Throwable e ) {
                 getLogger().error( "Error handling connection", e );
             } finally {
 
                 // Close the underlying socket
                 try {
                     if (clientSocket != null) {
                         clientSocket.close();
                     }
                 } catch( IOException ioe ) {
                     getLogger().warn( "Error shutting down connection", ioe );
                 }
 
                 clientSocket = null;
 
                 // Null out the thread, notify other threads to encourage
                 // a context switch
                 synchronized( this ) {
                     clientSocketThread = null;
 
                     Thread.currentThread().interrupted();
 
                     // Release the handler and kill the reference to the handler factory
                     //
                     // This needs to be done after the clientSocketThread is nulled out,
                     // otherwise we could trash a reused ClientConnectionRunner
                     if (handler != null) {
                         ServerConnection.this.handlerFactory.releaseConnectionHandler( handler );
                         handler = null;
                     }
 
                     // Remove this runner from the list of active connections.
                     ServerConnection.this.removeClientConnectionRunner(this);
 
                     notifyAll();
                 }
             }
         }
 
         /***
          * Helper method to return a formatted string with connection transport information.
          *
          * @return a formatted string
          */
         private String getConnectionString() {
             if (clientSocket == null) {
                 return "invalid socket";
             }
             StringBuffer connectionBuffer
                 = new StringBuffer(256)
                     .append("connection on ")
                     .append(clientSocket.getLocalAddress().getHostAddress().toString())
                     .append(":")
                     .append(clientSocket.getLocalPort())
                     .append(" from ")
                     .append(clientSocket.getInetAddress().getHostAddress().toString())
                     .append(":")
                     .append(clientSocket.getPort());
             return connectionBuffer.toString();
         }
     }
 
     /***
      * The factory for producing handlers.
      */
     private class ClientConnectionRunnerFactory
         implements ObjectFactory {
 
         /***
          * @see org.apache.avalon.excalibur.pool.ObjectFactory#newInstance()
          */
         public Object newInstance() throws Exception {
             return new ClientConnectionRunner();
         }
 
         /***
          * @see org.apache.avalon.excalibur.pool.ObjectFactory#getCreatedClass()
          */
         public Class getCreatedClass() {
             return ClientConnectionRunner.class;
         }
 
         /***
          * @see org.apache.avalon.excalibur.pool.ObjectFactory#decommision(Object)
          */
         public void decommission( Object object ) throws Exception {
             return;
         }
     }
 }