/**********************************************************************
    * ConnectionReader.java
    * created on 05.03.2005 by netseeker
    * $Source: /cvsroot/ejoe/EJOE/src/de/netseeker/ejoe/ConnectionReader.java,v $
    * $Date: 2006/03/22 00:06:58 $
    * $Revision: 1.9 $
    *
    * ====================================================================
    *
   *  Copyright 2005-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;
  
  import java.io.IOException;
  import java.io.InputStream;
  import java.nio.ByteBuffer;
  import java.nio.channels.Channels;
  import java.nio.channels.ClosedChannelException;
  import java.nio.channels.NonReadableChannelException;
  import java.nio.channels.ReadableByteChannel;
  import java.nio.channels.SelectionKey;
  import java.nio.channels.SocketChannel;
  import java.util.logging.Level;
  import java.util.logging.Logger;
  
  import de.netseeker.ejoe.adapter.AdapterFactory;
  import de.netseeker.ejoe.adapter.SerializeAdapter;
  import de.netseeker.ejoe.handler.ServerHandler;
  import de.netseeker.ejoe.io.ByteBufferInputStream;
  import de.netseeker.ejoe.io.ChannelInputStream;
  import de.netseeker.ejoe.io.IOUtils;
  import de.netseeker.ejoe.io.IncompleteIOException;
  
  /***
   * ConnectionReader targets two jobs:
   * <ul>
   * 	<li>Read (partial) data from a established client connection</li>
   *  <li>Invoke the server handler to process the received data</li>
   * </ul>
   * @author netseeker
   * @since 0.3.0
   */
  final class ConnectionReader implements Runnable
  {
  	private static final Logger		log	= Logger.getLogger(ConnectionReader.class.getName());
  
  	private final ChannelRegistrar	_registrar;
  
  	private final ServerHandler		_handler;
  
  	private ConnectionHeader		_senderInfo, _receiverInfo;
  
  	/***
  	 * @param channel a socket channel ready for reading and writing
  	 * @param adapter (de)serialize adapter
  	 * @param handler the working horse handling the transported input object
  	 */
  	public ConnectionReader(final ChannelRegistrar registrar, ConnectionHeader receiverInfo,
  			ConnectionHeader senderInfo, final ServerHandler handler)
  	{
  		this._registrar = registrar;
  		this._receiverInfo = receiverInfo;
  		this._senderInfo = senderInfo;
  		this._handler = handler;
  	}
  
  	/*
  	 * (non-Javadoc)
  	 *
  	 * @see java.lang.Runnable#run()
  	 */
  	public void run()
  	{
  		Object request = null;
  		SocketChannel channel = (SocketChannel) this._senderInfo.getChannel();
  
  		try
  		{
  			if (this._senderInfo == null || !this._senderInfo.isConnected())
  			{
  				log.log(Level.FINEST, "Handshaking client...");
 				// handshake the client, get infos about compression...
 				this._senderInfo = IOUtils.handshake(channel, this._receiverInfo, false,
 						EJConstants.EJOE_CONNECTION_TIMEOUT);
 			}
 
 			if (this._senderInfo != null)
 			{
 				log.log(Level.FINEST, "Client requested " + this._senderInfo.getAdapterName() );
 				SerializeAdapter adapter = AdapterFactory.createAdapter(this._senderInfo.getAdapterName());
 
 				if (this._receiverInfo.hasNonBlockingReadWrite())
 				{
 					log.log(Level.FINE, "Going to read client request on a non blocking socket...");
 					request = read(adapter);
 				}
 				else
 				{
 					log.log(Level.FINE, "Going to read client request on a blocking socket...");
 					request = readBlocked(adapter);
 				}
 
 				this._senderInfo.releaseWaitingBuffer();
 
 				log.log(Level.FINE, "Client request read.");
 				if (request != null)
 				{
 					Object result = handleObject(request);
 					if (this._registrar.isValid())
 					{
 						this._senderInfo.setAttachment(result);
 						this._registrar.register(this._senderInfo, SelectionKey.OP_WRITE);
 					}
 				}
 			}
 			else
 			{
 				log.log(Level.WARNING,
 						"Connection timeout reached while waiting for Handshake complete. Closing connection.");
 				IOUtils.closeQuite(channel);
 				this._senderInfo = null;
 			}
 		}
 		catch (IncompleteIOException ioe)
 		{
 			this._senderInfo.setWaitingBuffer(ioe.getIOBuffer());
 			this._registrar.register(this._senderInfo, SelectionKey.OP_READ);
 		}
 		catch (NonReadableChannelException e)
 		{
 			log.log(Level.INFO, "Connection closed by client.");
 			IOUtils.closeQuite(channel);
 			this._senderInfo = null;
 		}
 		catch (ClosedChannelException cce)
 		{
 			log.log(Level.INFO, "Connection closed by client.");
 			this._senderInfo = null;
 			IOUtils.closeQuite(channel);
 		}
 		catch (IOException e)
 		{
 			if (channel.isConnected() && channel.isOpen())
 			{
 				// something goes completely wrong!
 				log.log(Level.SEVERE, "!!! Exception while reading client data !!!", e);
 			}
 			else
 			{
 				log.log(Level.INFO, "Connection closed by client.");
 			}
 			IOUtils.closeQuite(channel);
 			this._senderInfo = null;
 		}
 		catch (Exception e)
 		{
 			// OK, we got screwed up completely
 			// something goes completely wrong!
 			log.log(Level.SEVERE, "!!! Exception while reading client data !!!", e);
 			IOUtils.closeQuite(channel);
 			this._senderInfo = null;
 		}
 	}
 
 	/***
 	 * Reads and deserialize a request object from a socket channel
 	 * @param adapter the deserialization adapter to use
 	 * @return the deserialized request object
 	 * @throws IOException
 	 */
 	private Object read(SerializeAdapter adapter) throws IOException
 	{
 		ByteBufferInputStream in = null;
 		ByteBuffer dataBuf = null;
 		Object result = null;
 		SocketChannel channel = (SocketChannel) this._senderInfo.getChannel();
 
 		try
 		{
 			if (!this._senderInfo.hasWaitingBuffer())
 			{
 				int length = IOUtils.readHeader(this._senderInfo);
 				dataBuf = ByteBuffer.allocateDirect(length);
 				log.log(Level.FINE, "Going to read client request with length: " + length);
 			}
 			else
 			{
 				dataBuf = this._senderInfo.getWaitingBuffer();
 			}
 
 			IOUtils.nonBlockingRead(channel, dataBuf);
 			dataBuf.flip();
 
 			in = new ByteBufferInputStream(dataBuf);
 			result = deserialize(adapter, in);
 		}
 		finally
 		{
 			IOUtils.closeQuite(in);
 		}
 
 		return result;
 	}
 
 	/***
 	 * @return
 	 * @throws IOException
 	 */
 	private Object readBlocked(SerializeAdapter adapter) throws IOException
 	{
 		return deserialize(adapter, new ChannelInputStream(Channels
 				.newInputStream((ReadableByteChannel) this._senderInfo.getChannel())));
 	}
 
 	/***
 	 * @param clientInfo
 	 * @param in
 	 * @return
 	 * @throws IOException
 	 */
 	private Object deserialize(SerializeAdapter adapter, InputStream in) throws IOException
 	{
 		boolean compressed = this._receiverInfo.hasCompression() && this._senderInfo.hasCompression();
 		return IOUtils.adapterDeserialize(adapter, in, compressed);
 	}
 
 	/***
 	 * @param obj
 	 * @return
 	 */
 	private Object handleObject(Object obj)
 	{
 		Object result = null;
 
 		try
 		{
 			result = _handler.handle(obj);
 		}
 		catch (Exception e)
 		{
 			log.log(Level.WARNING, "Exception in ServerHandler " + _handler + " occured.", e);
 			result = e;
 		}
 
 		return result;
 	}
 }