/**********************************************************************
    * CombinedConnectionProcessor.java
    * created on 01.03.2005 by netseeker
    * $Source: /cvsroot/ejoe/EJOE/src/de/netseeker/ejoe/core/CombinedConnectionProcessor.java,v $
    * $Date: 2007/03/22 21:01:28 $
    * $Revision: 1.3 $
    *
    * ====================================================================
    *
   *  Copyright 2006 netseeker aka Michael Manske
   *
   *  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.
   * ====================================================================
   *
   * This file is part of the ejoe framework.
   * For more information on the author, please see
   * <http://www.manskes.de/>.
   *
   *********************************************************************/
  package de.netseeker.ejoe.core;
  
  import java.io.IOException;
  import java.nio.channels.CancelledKeyException;
  import java.nio.channels.SelectionKey;
  import java.nio.channels.Selector;
  import java.nio.channels.SocketChannel;
  import java.util.ArrayList;
  import java.util.Collections;
  import java.util.Iterator;
  import java.util.List;
  import java.util.Set;
  import java.util.Timer;
  import java.util.logging.Level;
  import java.util.logging.Logger;
  
  import de.netseeker.ejoe.ConnectionHeader;
  import de.netseeker.ejoe.EJConstants;
  import de.netseeker.ejoe.ServerInfo;
  import de.netseeker.ejoe.concurrent.ThreadPoolFactory;
  import de.netseeker.ejoe.concurrent.ThreadPoolResizer;
  import de.netseeker.ejoe.concurrent.ThreadService;
  import de.netseeker.ejoe.io.IOUtil;
  
  /***
   * The CombinedConnectionProcessor handles the further processing of accepted client connections. It simply does only
   * separate the readable connections from the incoming flood of accepted connections and schedules a new
   * ConnectionProcessor which will handle read/write operations and adapter invoking for each of the selected
   * connections. This is compareable to the *old BIO style* where one thread is used for each connection.
   * 
   * @author netseeker
   * @since 0.3.5
   */
  public final class CombinedConnectionProcessor extends Thread implements ChannelRegistrar
  {
      private static final Logger  logger       = Logger.getLogger( ConnectionAcceptor.class.getName() );
  
      private static final Integer INT_OP_WRITE = new Integer( SelectionKey.OP_WRITE );
  
      private static final Integer INT_OP_READ  = new Integer( SelectionKey.OP_READ );
  
      private ServerInfo           _serverInfo;
  
      private ThreadGroup          _readGroup, _writeGroup;
  
      private ThreadService        _readpool, _writePool;
  
      private Selector             _selector;
  
      private List                 _aspirants;
  
      private int                  _load        = 0;
  
      /***
       * Creates a new CombinedConnectionProcessor instance. Using this constructor will force the newly created instance
       * to use a non-blocking threadpool for invocation of read processor threads.
       * 
       * @param serverInfo prefilled ConnectionHeader container server settings, eg. compression setting
       * @throws IOException
       */
      public CombinedConnectionProcessor(ServerInfo serverInfo) throws IOException
      {
          super( "EJOE CombinedConnectionProcessor" );
          this._serverInfo = serverInfo;
          // use our own ThreadGroups to simplify identifying EJOE threads in the
          // java (or systems) process list
          this._readGroup = new ThreadGroup( "EJOE_RP_THREADS" );
          this._writeGroup = new ThreadGroup( "EJOE_W_THREADS" );
  
          this._aspirants = Collections.synchronizedList( new ArrayList( 64 ) );
 
         // open a global Selector instance
         this._selector = Selector.open();
 
         // clean out previously cancelled keys - system might reuse selectors
         this._selector.selectNow();
     }
 
     /*
      * (non-Javadoc)
      * 
      * @see java.lang.Thread#run()
      */
     public void run()
     {
         Timer timer = null;
 
         this._readpool = ThreadPoolFactory.createFixedThreadPool( this._serverInfo.getMaxReadProcessors(),
                                                                   this._readGroup );
         // create a global ThreadPool for network write operations
         this._writePool = ThreadPoolFactory.createFixedThreadPool( this._serverInfo.getMaxWriteProcessors(),
                                                                    this._writeGroup );
 
         // shall we monitor the ThreadPools and resize them
         // according to server load (this will also recreate died worker
         // threads)
         if ( this._serverInfo.isAutomaticThreadPoolResize() )
         {
             timer = new Timer();
 
             // schedule a monitor task for the read-process-thread-pool
             timer.schedule( new ThreadPoolResizer( this._readpool, this._serverInfo.getMaxReadProcessors(),
                                                    EJConstants.EJOE_POOL_RESIZER_SHRINKWAIT ), this._serverInfo
                     .getPoolResizePeriod(), this._serverInfo.getPoolResizePeriod() );
             // schedule a monitor task for the writer-thread-pool
             timer.schedule( new ThreadPoolResizer( this._writePool, this._serverInfo.getMaxWriteProcessors(),
                                                    EJConstants.EJOE_POOL_RESIZER_SHRINKWAIT ), this._serverInfo
                     .getPoolResizePeriod(), this._serverInfo.getPoolResizePeriod() );
         }
 
         try
         {
             Iterator it;
             SocketChannel cChannel;
             SelectionKey selKey;
             ConnectionHeader clientInfo;
             Set keys;
 
             while ( !isInterrupted() )
             {
                 this._load = 0;
 
                 // (pre)register interested socket channels
                 registerAspirants();
 
                 // just try endless new selects until there are interested
                 // socket channels
                 if ( _selector.select() == 0 ) continue;
 
                 keys = this._selector.selectedKeys();
                 this._load = keys.size();
                 it = this._selector.selectedKeys().iterator();
 
                 // loop over all selected channels, just take care of thread
                 // interruption
                 while ( it.hasNext() && !isInterrupted() )
                 {
                     selKey = (SelectionKey) it.next();
                     // remove the SelectionKey from the Iterator otherwise it
                     // will be lost
                     it.remove();
 
                     try
                     {
                         // validate the key
                         if ( !selKey.isValid() ) continue;
 
                         // at least our ConnectionAcceptor has created a client
                         // ConnectionHeader
                         clientInfo = (ConnectionHeader) selKey.attachment();
 
                         // first check read-availbility
                         if ( selKey.isReadable() )
                         {
                             // get the underlying socket channel
                             cChannel = (SocketChannel) selKey.channel();
                             // cancel the channels registration with our
                             // Selector
                             selKey.cancel();
 
                             // little bit paranoia
                             if ( cChannel != null && cChannel.isOpen() )
                             {
                                 // don't we support NIO?
                                 if ( !this._serverInfo.hasNonBlockingReadWrite() )
                                 {
                                     logger.log( Level.FINEST,
                                                 "Setting socket to blocking mode for further io operations..." );
 
                                     // prepare the channel for upcoming blocking
                                     // network operations
                                     cChannel.configureBlocking( true );
                                 }
 
                                 // schedule a asynchronious read-process operation
                                 this._readpool.invokeLater( new ConnectionReader( this, this._serverInfo, clientInfo ) );
                             }
                         }
                         else if ( selKey.isWritable() )
                         {
                             // get the underlying socket channel
                             cChannel = (SocketChannel) selKey.channel();
                             // cancel the channels registration with our
                             // Selector
                             selKey.cancel();
 
                             // little bit paranoia
                             if ( cChannel != null && cChannel.isOpen() )
                             {
                                 // don't we support NIO?
                                 if ( !this._serverInfo.hasNonBlockingReadWrite() )
                                 {
                                     if ( !clientInfo.hasAttachment() ) continue;
                                     // prepare the channel for upcoming blocking
                                     // network operations
                                     cChannel.configureBlocking( true );
                                 }
 
                                 // schedule a asynchronious socket-write
                                 // operation
                                 this._writePool
                                         .invokeLater( new ConnectionWriter( this, this._serverInfo, clientInfo ) );
                             }
                         }
                     }
                     catch ( CancelledKeyException cke )
                     {
                         logger.log( Level.WARNING, "Key cancelled!", cke );
                     }
                     finally
                     {
                         this._load--;
                     }
                 }
             }
         }
         catch ( IOException e )
         {
             logger.log( Level.SEVERE, "!!! IOException occured !!! ", e );
             throw new RuntimeException( e );
         }
         finally
         {
             // kill our ThreadPool monitors if existent
             if ( timer != null ) timer.cancel();
 
             try
             {
                 // shutdown the Read-Process-ThreadPool if existent
                 if ( this._readpool != null )
                 {
                     this._readpool.stop();
                 }
             }
             catch ( Exception e )
             {
                 logger.log( Level.SEVERE, "!!! Error while stopping server !!!", e );
             }
             try
             {
                 // shutdown the Write-ThreadPool
                 this._writePool.stop();
             }
             catch ( Exception e )
             {
                 logger.log( Level.SEVERE, "!!! Error while stopping server !!!", e );
             }
 
             // and finally close the global Selector
             IOUtil.closeQuiet( this._selector );
         }
     }
 
     /***
      * Registers all temporalily queued socket channels on the used {@link Selector}
      * 
      * @throws IOException
      */
     private void registerAspirants() throws IOException
     {
         int count = 0;
         int size = this._aspirants.size();
 
         if ( size > 0 )
         {
             // clean out cancelled key list
             this._selector.selectNow();
             Object[] arr = null;
             ConnectionHeader header = null;
             SocketChannel channel = null;
             SelectionKey sk = null;
 
             for ( ; count < size; count++ )
             {
                 arr = (Object[]) this._aspirants.remove( 0 );
                 header = (ConnectionHeader) arr[0];
                 channel = header.getChannel();
                 // set a previously blocking socket channel to non-blocking
                 // mode to allow selector operations on it
                 if ( channel.isBlocking() )
                 {
                     logger.log( Level.FINEST,
                                 "Setting socket temporarily to non-blocking until further connection processing..." );
                     channel.configureBlocking( false );
                 }
 
                 sk = channel.keyFor( this._selector );
                 if ( sk == null )
                 {
                     // register with no interest
                     sk = channel.register( this._selector, 0 );
                 }
 
                 // attach the client connection header
                 sk.attach( header );
                 // now set the requested interest
                 sk.interestOps( ((Integer) arr[1]).intValue() );
             }
         }
 
         if ( logger.isLoggable( Level.FINEST ) )
         {
             logger.log( Level.FINEST, System.currentTimeMillis() + " - registered " + count + " aspirants..." );
         }
     }
 
     /*
      * (non-Javadoc)
      * 
      * @see de.netseeker.ejoe.io.ReadWriteChannelRegistrar#register(de.netseeker.ejoe.ConnectionHeader, int)
      */
     public void register( ConnectionHeader clientInfo, int interest )
     {
         // store the connection header with it's contained socket channel in
         // the global queue
         this._aspirants
                 .add( new Object[] { clientInfo, (interest == SelectionKey.OP_READ) ? INT_OP_READ : INT_OP_WRITE } );
 
         // break blocking select operations in the selector to allow
         // registration of the socket channels within the registration queue
         // after processing all currently selected socket channels
         this._selector.wakeup();
     }
 
     /*
      * (non-Javadoc)
      * 
      * @see de.netseeker.ejoe.ChannelRegistrar#isValid()
      */
     public boolean isValid()
     {
         return isAlive() && !isInterrupted();
     }
 
     /*
      * (non-Javadoc)
      * 
      * @see de.netseeker.ejoe.core.ChannelRegistrar#getLoad()
      */
     public int getLoad()
     {
         return this._load;
     }
 }