/**********************************************************************
    * ConnectionProcessorManager.java
    * created on 01.03.2005 by netseeker
    * $Source: /cvsroot/ejoe/EJOE/src/de/netseeker/ejoe/ConnectionProcessorManager.java,v $
    * $Date: 2006/02/04 16:16:48 $
    * $Revision: 1.29 $
    *********************************************************************/
   package de.netseeker.ejoe;
   
  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.nio.channels.spi.SelectorProvider;
  import java.util.Iterator;
  import java.util.LinkedHashMap;
  import java.util.Map;
  import java.util.logging.Level;
  import java.util.logging.Logger;
  
  import de.netseeker.ejoe.concurrent.ThreadPool;
  import de.netseeker.ejoe.handler.ServerHandler;
  import de.netseeker.ejoe.io.IOUtils;
  
  /***
   * The ConnectionProcessorManager 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
   */
  final class ConnectionProcessorManager extends Thread implements ChannelRegistrar
  {
  	private static final Logger	logger					= Logger.getLogger(ConnectionAcceptor.class.getName());
  
  	private ConnectionHeader	_serverInfo;
  
  	private final ServerHandler	_handler;
  
  	private final int			_maxReadProcessors, _maxWriteProcessors;
  
  	private boolean				_useThreadedProcessors	= false;
  
  	private ThreadGroup			_readGroup, _writeGroup;
  
  	private ThreadPool			_readpool, _writePool;
  
  	private Selector			_selector;
  
  	private Map					_aspirants;
  
  	/***
  	 * Creates a new ConnectionProcessorManager 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
  	 * @param handler
  	 *            server handler to use when processing retrieved data
  	 * @param adapter
  	 *            (de)serialize adapter to use when deserializing data during read or rather serializing data during
  	 *            write operations
  	 * @param maxReadProcessors
  	 *            amount of read processors threads to use for reading data and processing these retrieved data
  	 * @param maxWriteProcessors
  	 *            amount of writer threads to use for sending server responses to clients
  	 * @throws IOException
  	 */
  	public ConnectionProcessorManager(ConnectionHeader serverInfo, final ServerHandler handler, int maxReadProcessors,
  			int maxWriteProcessors) throws IOException
  	{
  		this._serverInfo = serverInfo;
  		this._handler = handler;
  		this._maxReadProcessors = maxReadProcessors;
  		this._maxWriteProcessors = maxWriteProcessors;
  		this._readGroup = new ThreadGroup("EJOE_RP_THREADS");
  		this._writeGroup = new ThreadGroup("EJOE_W_THREADS");
  		this._selector = SelectorProvider.provider().openSelector();
  		this._aspirants = new LinkedHashMap(Math.round((float) ((maxReadProcessors + maxWriteProcessors) * 1.5)));
  	}
  
  	/***
  	 * Creates a new ConnectionProcessorManager instance
  	 *
  	 * @param serverInfo
  	 *            prefilled ConnectionHeader container server settings, eg. compression setting
  	 * @param handler
  	 *            server handler to use when processing retrieved data
  	 * @param adapter
  	 *            (de)serialize adapter to use when deserializing data during read or rather serializing data during
  	 *            write operations
  	 * @param maxReadProcessors
  	 *            amount of read processors threads to use for reading data and processing these retrieved data
  	 * @param maxWriteProcessors
  	 *            amount of writer threads to use for sending server responses to clients
  	 * @param useThreadedReadProcessors
  	 *            whether to use a blocking thread start mechanism, which will invoke new read processor threads not
 	 *            until the amount of concurrently running read processors is lower than <var>maxReadProcessors </var>
 	 * @throws IOException
 	 */
 	public ConnectionProcessorManager(ConnectionHeader serverInfo, final ServerHandler handler, int maxProcessors,
 			int maxWriteProcessors, boolean useThreadedReadProcessors) throws IOException
 	{
 		this(serverInfo, handler, maxProcessors, maxWriteProcessors);
 		this._useThreadedProcessors = useThreadedReadProcessors;
 	}
 
 	/*
 	 * (non-Javadoc)
 	 *
 	 * @see java.lang.Thread#run()
 	 */
 	public void run()
 	{
 		if (!this._useThreadedProcessors)
 		{
 			this._readpool = new ThreadPool(this._readGroup, this._maxReadProcessors);
 		}
 
 		this._writePool = new ThreadPool(this._writeGroup, this._maxWriteProcessors);
 
 		try
 		{
 			registerAspirants();
 
 			while (!isInterrupted())
 			{
 				this._selector.select();
 				registerAspirants();
 				Iterator it = this._selector.selectedKeys().iterator();
 				SocketChannel cChannel;
 
 				while (it.hasNext() && !isInterrupted())
 				{
 					SelectionKey selKey = (SelectionKey) it.next();
 					it.remove();
 
 					try
 					{
 						// validate the key
 						if (!selKey.isValid())
 						{
 							continue;
 						}
 
 						ConnectionHeader clientInfo = (ConnectionHeader) selKey.attachment();
 
 						if (selKey.isReadable())
 						{
 							cChannel = (SocketChannel) selKey.channel();
 							selKey.cancel();
 
 							if (cChannel != null)
 							{
 								if (!this._serverInfo.hasNonBlockingReadWrite())
 								{
 									logger.log(Level.FINEST,
 											"Setting socket to blocking mode for further io operations...");
 									cChannel.configureBlocking(true);
 								}
 
 								invokeReadOperation(new ConnectionReader(this, this._serverInfo, clientInfo,
 										this._handler));
 							}
 						}
 						else if (selKey.isWritable())
 						{
 							cChannel = (SocketChannel) selKey.channel();
 							selKey.cancel();
 
 							if (cChannel != null)
 							{
 								if (!this._serverInfo.hasNonBlockingReadWrite())
 								{
 									cChannel.configureBlocking(true);
 								}
 
 								this._writePool.invokeLater(new ConnectionWriter(this, this._serverInfo, clientInfo));
 							}
 						}
 					}
 					catch (CancelledKeyException cke)
 					{
 						logger.log(Level.WARNING, "Key cancelled!", cke);
 					}
 				}
 			}
 		}
 		catch (IOException e)
 		{
 			logger.log(Level.SEVERE, "!!! IOException occured !!! ", e);
 			throw new RuntimeException(e);
 		}
 		finally
 		{
 			try
 			{
 				if (this._readpool != null)
 				{
 					this._readpool.stop();
 				}
 			}
 			catch (Exception e)
 			{
 				logger.log(Level.SEVERE, "!!! Error while stopping server !!!", e);
 			}
 			try
 			{
 				this._writePool.stop();
 			}
 			catch (Exception e)
 			{
 				logger.log(Level.SEVERE, "!!! Error while stopping server !!!", e);
 			}
 
 			IOUtils.closeQuite(this._selector);
 		}
 	}
 
 	/***
 	 * @param runnable
 	 */
 	private void invokeReadOperation(Runnable runnable)
 	{
 		if (!isInterrupted())
 		{
 			if (!this._useThreadedProcessors)
 			{
 				this._readpool.invokeLater(runnable);
 			}
 			else
 			{
 				try
 				{
 					synchronized (this._readGroup)
 					{
 						while (this._readGroup.activeCount() >= this._maxReadProcessors)
 						{
 							this._readGroup.wait(EJConstants.EJOE_WAIT_TIMEOUT);
 						}
 					}
 				}
 				catch (InterruptedException e)
 				{
 					return;
 				}
 
 				new Thread(this._readGroup, runnable).start();
 			}
 		}
 	}
 
 	private void registerAspirants() throws IOException
 	{
 		int count = 0;
 
 		synchronized (this._aspirants)
 		{
 			if (!this._aspirants.isEmpty())
 			{
 				// clean out cancelled key list before (re-)registering
 				// new channels
 				this._selector.selectNow();
 				Iterator ait = this._aspirants.keySet().iterator();
 				ConnectionHeader header = null;
 				SocketChannel channel = null;
 				while (ait.hasNext())
 				{
 					header = (ConnectionHeader) ait.next();
 					channel = (SocketChannel)header.getChannel();
 					if (channel.isBlocking())
 					{
 						logger.log(Level.FINEST,
 								"Setting socket temporarily to non-blocking until further connection processing...");
 						channel.configureBlocking(false);
 					}
 					channel.register(this._selector, ((Integer) this._aspirants.get(header)).intValue(), header);
 					ait.remove();
 					count++;
 				}
 			}
 
 			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)
 	{
 		synchronized (this._aspirants)
 		{
 			this._aspirants.put(clientInfo, new Integer(interest));
 			this._selector.wakeup();
 		}
 	}
 
 	/*
 	 * (non-Javadoc)
 	 *
 	 * @see de.netseeker.ejoe.ChannelRegistrar#isValid()
 	 */
 	public boolean isValid()
 	{
 		return isAlive() && !isInterrupted();
 	}
 }