/**********************************************************************
    * IOUtils.java
    * created on 15.08.2004 by netseeker
    * $Source: /cvsroot/ejoe/EJOE/src/de/netseeker/ejoe/io/IOUtils.java,v $
    * $Date: 2006/02/04 14:15:53 $
    * $Revision: 1.32 $
    *********************************************************************/
   
   package de.netseeker.ejoe.io;
  
  import java.io.BufferedInputStream;
  import java.io.BufferedOutputStream;
  import java.io.IOException;
  import java.io.InputStream;
  import java.io.OutputStream;
  import java.io.Reader;
  import java.io.Writer;
  import java.net.Socket;
  import java.net.SocketException;
  import java.nio.ByteBuffer;
  import java.nio.channels.Channel;
  import java.nio.channels.ClosedChannelException;
  import java.nio.channels.ReadableByteChannel;
  import java.nio.channels.SelectionKey;
  import java.nio.channels.Selector;
  import java.nio.channels.SocketChannel;
  import java.nio.channels.WritableByteChannel;
  import java.util.logging.Level;
  import java.util.logging.Logger;
  import java.util.zip.DeflaterOutputStream;
  import java.util.zip.GZIPInputStream;
  import java.util.zip.GZIPOutputStream;
  
  import de.netseeker.ejoe.ConnectionHeader;
  import de.netseeker.ejoe.EJConstants;
  import de.netseeker.ejoe.adapter.SerializeAdapter;
  
  /***
   * Some useful methods for closing io streams, io readers, channels and Selector
   * quitely, as well as methods for non-blocking, semi-blocking and full blocking
   * io read/write operations.
   *
   * @author netseeker aka Michael Manske
   */
  public final class IOUtils
  {
  	public static final int		NIO_MAX_ITERATIONS	= 3;
  
  	private static final Logger	logger				= Logger.getLogger(IOUtils.class.getName());
  
  	/***
  	 * Tries to close an OutputStream and handles null values and IOExceptions
  	 * quietly
  	 *
  	 * @param out
  	 */
  	public static void closeQuite(OutputStream out)
  	{
  		if (out != null)
  		{
  			try
  			{
  				out.close();
  			}
  			catch (IOException e)
  			{
  				// do nothing
  			}
  		}
  	}
  
  	/***
  	 * Tries to close a Writer and handles null values and IOExceptions quietly
  	 *
  	 * @param out
  	 */
  	public static void closeQuite(Writer out)
  	{
  		if (out != null)
  		{
  			try
  			{
  				out.close();
  			}
  			catch (IOException e)
  			{
  				// do nothing
  			}
  		}
  	}
  
  	/***
  	 * Tries to close an InputStream and handles null values and IOExceptions
  	 * quietly
  	 *
  	 * @param in
  	 */
  	public static void closeQuite(InputStream in)
  	{
 		if (in != null)
 		{
 			try
 			{
 				in.close();
 			}
 			catch (IOException e)
 			{
 				// do nothing
 			}
 		}
 	}
 
 	/***
 	 * Tries to close a Reader and handles null values and IOExceptions quietly
 	 *
 	 * @param reader
 	 */
 	public static void closeQuite(Reader reader)
 	{
 		if (reader != null)
 		{
 			try
 			{
 				reader.close();
 			}
 			catch (IOException e)
 			{
 				// do nothing
 			}
 		}
 	}
 
 	/***
 	 * Tries to close an NIO Channel and handles null values and IOExceptions
 	 * quietly
 	 *
 	 * @param channel
 	 */
 	public static void closeQuite(Channel channel)
 	{
 		if (channel != null)
 		{
 			try
 			{
 				channel.close();
 			}
 			catch (IOException e)
 			{
 				// do nothing
 			}
 		}
 	}
 
 	/***
 	 * Tries to close a NIO Selector and handles null values and IOExceptions
 	 * quietly
 	 *
 	 * @param selector
 	 */
 	public static void closeQuite(Selector selector)
 	{
 		if (selector != null && selector.isOpen())
 		{
 			try
 			{
 				selector.close();
 			}
 			catch (IOException e)
 			{
 				// do nothing
 			}
 		}
 	}
 
 	/***
 	 * Handshake for a socket channel. It's used as workaround for a know issue
 	 * with java sockets: Sometimes only the first byte will get transferred
 	 * through a socket connection when reading from it first time. The other
 	 * bytes will follow not until the next read. This method sends/receives one
 	 * Byte through the socket to "initialize" the socket channel. So all
 	 * following read/write operations don't have to handle that "1-Byte issue".
 	 * The send/received Byte is used also as connection header, it contains
 	 * information about compression, nio usage, if the connection is a
 	 * persistent or non-persistent one...
 	 *
 	 * @param channel
 	 *            A connected, readable and writeable socket channel
 	 * @param sendBeforeReceive
 	 *            if true we will try to send one byte then read one byte
 	 *            otherwise we will use the opposite way around.
 	 * @throws IOException
 	 */
 	public static ConnectionHeader handshake(SocketChannel channel, final ConnectionHeader header,
 			boolean sendBeforeReceive, long timeout) throws IOException
 	{
 		ConnectionHeader receiverHeader;
 		ByteBuffer magicBuf = null;
 
 		try
 		{
 			// shall we act as clientside and initialize the handshake?
 			if (sendBeforeReceive)
 			{
 				// ok, get the control bits as well as the adapter name in a
 				// ByteChannel
 				magicBuf = header.toByteBuffer();
 				logger.log(Level.FINEST, "Sending Clientheader: " + header);
 				semiBlockingWrite(channel, magicBuf, timeout);
 				if (magicBuf.hasRemaining()) return null;
 				// the server will just answer with the header byte
 				magicBuf.clear();
 				magicBuf.limit(1);
 			}
 			else
 			{
 				// expect to read the header byte and an integer (the size of
 				// the forthcoming adapter name string)
 				magicBuf = ByteBuffer.allocateDirect(5);
 			}
 
 			// Server: read the client header byte + sizeof adapter name
 			// Client: read just the server header byte
 			semiBlockingRead(channel, magicBuf, timeout);
 			if (magicBuf.hasRemaining()) return null;
 			magicBuf.flip();
 			receiverHeader = new ConnectionHeader(channel, magicBuf.get());
 
 			// shall we act as server side and answer to the handshake request?
 			if (!sendBeforeReceive)
 			{
 				// ok, we have already read the client header byte, now read the
 				// size of the adapter name string
 				int length = magicBuf.getInt();
 				// client requested a special adapter?
 				if (length > 0)
 				{
 					// limit the existing ByteBuffer if it's big enough
 					if (length <= magicBuf.capacity())
 					{
 						magicBuf.clear();
 						magicBuf.limit(length);
 					}
 					// otherwise allocate a new one
 					else
 					{
 						magicBuf = ByteBuffer.allocateDirect(length);
 					}
 
 					// read the adapter name
 					semiBlockingRead(channel, magicBuf, timeout);
 					if (magicBuf.hasRemaining()) return null;
 					magicBuf.flip();
 
 					byte[] adapterArr = new byte[length];
 					magicBuf.get(adapterArr);
 					receiverHeader.setAdapterName(new String(adapterArr, "UTF8"));
 				}
 
 				// answer to the client request and send our serverside header
 				// byte
 				magicBuf.limit(1);
 				magicBuf.clear();
 				magicBuf.put(header.toByte());
 				magicBuf.flip();
 				semiBlockingWrite(channel, magicBuf, timeout);
 				if (magicBuf.hasRemaining()) return null;
 			}
 
 			receiverHeader.setConnected(true);
 		}
 		finally
 		{
 			magicBuf = null;
 		}
 
 		return receiverHeader;
 	}
 
 	/***
 	 * Invokes a SerializeAdapter for the given Object and handles compression
 	 * and buffering
 	 *
 	 * @param adapter
 	 * @param out
 	 * @param obj
 	 * @param buffered
 	 * @param compressed
 	 * @throws IOException
 	 */
 	public static void adapterSerialize(final SerializeAdapter adapter, OutputStream out, Object obj,
 			boolean compressed, final int compressionLevel) throws IOException
 	{
 		logger.log(Level.FINEST, "Using compression: " + compressed);
 
 		if (compressed)
 		{
 			DeflaterOutputStream sOut = new GZIPOutputStream(out, EJConstants.BUFFERED_STREAM_SIZE)
 				{
 					{
 						def.setLevel(compressionLevel);
 					}
 				};
 			adapter.write(obj, sOut);
 			sOut.finish();
 		}
 		else
 		{
 			BufferedOutputStream sOut = new BufferedOutputStream(out, EJConstants.BUFFERED_STREAM_SIZE);
 			adapter.write(obj, sOut);
 			sOut.flush();
 		}
 	}
 
 	/***
 	 * Invokes a (De)SerializeAdapter for the given Object and handles
 	 * decompression and buffering
 	 *
 	 * @param adapter
 	 * @param in
 	 * @param buffered
 	 * @param compressed
 	 * @return
 	 * @throws IOException
 	 */
 	public static Object adapterDeserialize(final SerializeAdapter adapter, InputStream in, boolean compressed)
 			throws IOException
 	{
 		InputStream sIn = null;
 
 		logger.log(Level.FINEST, "Using compression: " + compressed);
 
 		if (compressed)
 		{
 			sIn = new GZIPInputStream(in, EJConstants.BUFFERED_STREAM_SIZE);
 		}
 		else
 		{
 			sIn = new BufferedInputStream(in, EJConstants.BUFFERED_STREAM_SIZE);
 		}
 
 		return adapter.read(sIn);
 	}
 
 	/***
 	 * Tries to send the given ByteBuffer completely through the given
 	 * SocketChannel three times
 	 *
 	 * @param channel
 	 * @param buffer
 	 * @throws IncompleteIOException if the given ByteBuffer could not be send completely
 	 * @throws IOException
 	 */
 	public static void nonBlockingWrite(WritableByteChannel channel, ByteBuffer buffer) throws IOException
 	{
 		int runs = 0;
 		//setSendBufferSize(channel.socket(), buffer.remaining());
 
 		do
 		{
 			channel.write(buffer);
 			runs++;
 		}
 		while (buffer.hasRemaining() && runs < EJConstants.NIO_MAX_ITERATIONS);
 
 		if (buffer.hasRemaining())
 		{
 			logger.log(Level.FINEST, "Incomplete write detected, registering for write again.");
 			throw new IncompleteIOException(buffer, SelectionKey.OP_WRITE);
 		}
 	}
 
 	/***
 	 * Tries to send the given ByteBuffer completely through the given
 	 * SocketChannel within a given timeout
 	 *
 	 * @param channel
 	 * @param buffer
 	 * @param timeout
 	 * @throws IncompleteIOException if the given ByteBuffer could not be send completely
 	 * @throws IOException
 	 */
 	public static void semiBlockingWrite(WritableByteChannel channel, ByteBuffer buffer, long timeout)
 			throws IOException
 	{
 		long timestamp = System.currentTimeMillis();
 		//setSendBufferSize(channel.socket(), buffer.remaining());
 
 		do
 		{
 			channel.write(buffer);
 		}
 		while (buffer.hasRemaining() && ((System.currentTimeMillis() - timestamp) < timeout));
 	}
 
 	/***
 	 * Tries to send the given ByteBuffer completely through the given
 	 * SocketChannel within a given timeout
 	 *
 	 * @param channel
 	 * @param buffer
 	 * @return
 	 * @throws IOException
 	 */
 	public static ByteBuffer nonBlockingRead(ReadableByteChannel channel, ByteBuffer buffer) throws IOException
 	{
 		int read = 0, runs = 0;
 
 		try
 		{
 			//setReceiveBufferSize(channel.socket(), buffer.remaining());
 
 			do
 			{
 				read = channel.read(buffer);
 				runs++;
 			}
 			while (buffer.hasRemaining() && runs < EJConstants.NIO_MAX_ITERATIONS && read != -1);
 		}
 		catch (IOException e)
 		{
 			// most likely the sender did close the connection
 			throw new ClosedChannelException();
 		}
 
 		if (buffer.hasRemaining())
 		{
 			if (read == -1)
 			{
 				throw new ClosedChannelException();
 			}
 			else
 			{
 				logger.log(Level.FINEST, "Incomplete read detected, registering for read again.");
 				throw new IncompleteIOException(buffer, SelectionKey.OP_READ);
 			}
 		}
 
 		return buffer;
 	}
 
 	/***
 	 * Tries to read ByteBuffer.remaining() bytes the into given ByteBuffer from
 	 * the given SocketChannel within a given timeout
 	 *
 	 * @param channel
 	 * @param buffer
 	 * @param timeout
 	 * @return
 	 * @throws IOException
 	 */
 	public static ByteBuffer semiBlockingRead(ReadableByteChannel channel, ByteBuffer buffer, long timeout)
 			throws IOException
 	{
 		long timestamp = System.currentTimeMillis();
 		int read = 0;
 
 		try
 		{
 			//setReceiveBufferSize(channel.socket(), buffer.remaining());
 
 			do
 			{
 				read = channel.read(buffer);
 			}
 			while (read != -1 && buffer.hasRemaining() && ((System.currentTimeMillis() - timestamp) < timeout));
 		}
 		catch (IOException e)
 		{
 			// most likely the sender did close the connection
 			throw new ClosedChannelException();
 		}
 
 		if (read == -1 && buffer.hasRemaining())
 		{
 			//either the client is too slow for us or the connection was dropped
 			throw new ClosedChannelException();
 		}
 
 		return buffer;
 	}
 
 	/***
 	 * Receives a EJOE specific header containing the size of the next
 	 * ByteBuffer.
 	 *
 	 * @param channel
 	 * @return the length of the following data package
 	 * @throws IOException
 	 */
 	public static int readHeader(ConnectionHeader header) throws IOException
 	{
 		ByteBuffer headerBuf = ByteBuffer.allocate(EJConstants.NIO_HEADER_SIZE);
 		int read = 0;
 		SocketChannel channel = (SocketChannel) header.getChannel();
 
 		try
 		{
 			nonBlockingRead(channel, headerBuf);
 			headerBuf.flip();
 			read = headerBuf.getInt();
 			setReceiveBufferSize(channel.socket(), read);
 		}
 		catch (IncompleteIOException ioe)
 		{
 			logger.log(Level.FINEST, "Incomplete header read detected, registering for read again.");
 			//ioe.setIOBuffer(null);
 			//ioe.setSelectionInterest(SelectionKey.OP_READ);
 			throw new IncompleteIOException(null, SelectionKey.OP_READ);
 		}
 		finally
 		{
 			headerBuf = null;
 		}
 
 		return read;
 	}
 
 	/***
 	 * Sends a EJOE specific header containing the lengh of the given ByteBuffer
 	 *
 	 * @param channel
 	 * @param buffer
 	 * @throws IOException
 	 */
 	public static void writeHeader(ConnectionHeader header, ByteBuffer buffer) throws IOException
 	{
 		ByteBuffer headerBuf = ByteBuffer.allocate(EJConstants.NIO_HEADER_SIZE);
 		SocketChannel channel = (SocketChannel) header.getChannel();
 
 		int length = buffer != null ? buffer.remaining() : 0;
 		headerBuf.putInt(length);
 		headerBuf.flip();
 		try
 		{
 			nonBlockingWrite(channel, headerBuf);
 			setSendBufferSize(channel.socket(), length);
 		}
 		catch (IncompleteIOException ioe)
 		{
 			logger.log(Level.FINEST, "Incomplete header write detected, registering for write again.");
 			//ioe.setIOBuffer(null);
 			//ioe.setSelectionInterest(SelectionKey.OP_WRITE);
 			throw new IncompleteIOException(null, SelectionKey.OP_WRITE);
 		}
 		finally
 		{
 			headerBuf = null;
 		}
 	}
 
 	/***
 	 * Set the SO_SNDBUF hint on a connected socket to the size of the data which
 	 * are expected to be written next time
 	 * @param socket the connected socket
 	 * @param size size to set for SO_SNDBUF
 	 * @throws SocketException
 	 */
 	public static void setSendBufferSize(Socket socket, int size) throws SocketException
 	{
 		if (size > 0)
 		{
 			if (size <= EJConstants.EJOE_MAX_SOCKET_BUF_SIZE)
 			{
 				socket.setSendBufferSize(size);
 			}
 			else
 			{
 				socket.setSendBufferSize(EJConstants.EJOE_MAX_SOCKET_BUF_SIZE);
 			}
 		}
 	}
 
 	/***
 	 * Set the SO_RCVBUF hint on a connected socket to the size of the data which
 	 * are expected to be read next time
 	 * @param socket the connected socket
 	 * @param size size to set for SO_RCVBUF
 	 * @throws SocketException
 	 */
 	public static void setReceiveBufferSize(Socket socket, int size) throws SocketException
 	{
 		if (size > 0)
 		{
 			if (size <= EJConstants.EJOE_MAX_SOCKET_BUF_SIZE)
 			{
 				socket.setReceiveBufferSize(size);
 			}
 			else
 			{
 				socket.setReceiveBufferSize(EJConstants.EJOE_MAX_SOCKET_BUF_SIZE);
 			}
 		}
 	}
 }