/**********************************************************************
    * EJClient.java
    * created on 08.08.2004 by netseeker
    * $Source: /cvsroot/ejoe/EJOE/src/de/netseeker/ejoe/EJClient.java,v $
    * $Date: 2006/02/04 15:17:45 $
    * $Revision: 1.58 $
    *********************************************************************/
   
   package de.netseeker.ejoe;
  
  import java.io.FileInputStream;
  import java.io.IOException;
  import java.io.InputStream;
  import java.io.OutputStream;
  import java.io.Serializable;
  import java.net.InetSocketAddress;
  import java.nio.ByteBuffer;
  import java.nio.channels.Channels;
  import java.nio.channels.SelectionKey;
  import java.nio.channels.Selector;
  import java.nio.channels.SocketChannel;
  import java.rmi.RemoteException;
  import java.util.Iterator;
  import java.util.Properties;
  import java.util.logging.Level;
  import java.util.logging.Logger;
  
  import de.netseeker.ejoe.adapter.ObjectStreamAdapter;
  import de.netseeker.ejoe.adapter.SerializeAdapter;
  import de.netseeker.ejoe.adapter.XStreamAdapter;
  import de.netseeker.ejoe.io.ByteBufferInputStream;
  import de.netseeker.ejoe.io.ByteBufferOutputStream;
  import de.netseeker.ejoe.io.IOUtils;
  import de.netseeker.ejoe.io.IncompleteIOException;
  
  /***
   * This is the client component of EJOE. You have to use this component to send and retrieve data to/from a EJOE server.
   *
   * @author netseeker aka Michael Manske
   */
  public class EJClient implements Serializable
  {
  	private static final long		serialVersionUID	= 3258413932539359280L;
  
  	private static final Logger		log					= Logger.getLogger(EJClient.class.getName());
  
  	private SerializeAdapter		_adapter;
  
  	private String					_host				= "127.0.0.1";
  
  	private int						_port				= EJConstants.EJOE_PORT;
  
  	private int						_classLoaderPort	= EJConstants.EJOE_CLASSLOADER_PORT;
  
  	private int						_connectionTimeout	= EJConstants.EJOE_CONNECTION_TIMEOUT;
  
  	private final ConnectionHeader	_clientInfo			= new ConnectionHeader();
  
  	private SocketChannel			_channel;
  
  	private boolean					_requestInProgress	= false;
  
  	/***
  	 * Creates an instance of the EJOE client pre-configured with settings from the global ejoe.properties file. You
  	 * MUST provide such an property file to use this constructor.
  	 */
  	public EJClient()
  	{
  		Properties props = new Properties();
  		try
  		{
  			props.load(EJClient.class.getResourceAsStream("/ejoe.properties"));
  			loadProperties(props);
  		}
  		catch (IOException e)
  		{
  			log
  					.log(
  							Level.SEVERE,
  							"ejoe.properties could not be read! Make sure you have placed a valid properties file in your classpath.",
  							e);
  			throw new RuntimeException(e);
  		}
  	}
  
  	/***
  	 * Creates an instance of the EJOE client pre-configured with settings from a global properties file.
  	 *
  	 * @param pathToConfigFile
  	 *            path to the properties file
  	 */
  	public EJClient(String pathToConfigFile)
  	{
  		Properties props = new Properties();
  		try
  		{
  			props.load(new FileInputStream(pathToConfigFile));
  			loadProperties(props);
  		}
 		catch (IOException e)
 		{
 			log
 					.log(
 							Level.SEVERE,
 							"ejoe.properties could not be read! Make sure you have placed a valid properties file in your classpath.",
 							e);
 			throw new RuntimeException(e);
 		}
 	}
 
 	/***
 	 * Creates an instance of the EJOE client preconfigured to use an instance of
 	 * de.netseeker.ejoe.adapter.ObjectStreamAdapter for (de)serializing.
 	 *
 	 * @param host
 	 *            address (dns name or ip address) of the EJOE server
 	 * @param port
 	 *            port which the EJOE server listens to
 	 */
 	public EJClient(String host, int port)
 	{
 		this._host = host;
 		this._port = port;
 		this._adapter = new XStreamAdapter();
 		this._clientInfo.setAdapterName(XStreamAdapter.class.getName());
 	}
 
 	/***
 	 * Creates an instance of the EJOE client.
 	 *
 	 * @param host
 	 *            address (dns name or ip address) of the EJOE server
 	 * @param port
 	 *            port which the EJOE server listens to
 	 * @param adapter
 	 *            the adapter used for (de)serializing input paramter objects for the server and the return values
 	 */
 	public EJClient(String host, int port, final SerializeAdapter adapter)
 	{
 		this._host = host;
 		this._port = port;
 		this._adapter = adapter;
 		this._clientInfo.setAdapterName(adapter.getClass().getName());
 	}
 
 	/***
 	 * Creates an instance of the EJOE client with remote classloading enabled.
 	 *
 	 * @param host
 	 *            address (dns name or ip address) of the EJOE server
 	 * @param port
 	 *            port which the EJOE server listens to
 	 * @param classLoaderPort
 	 *            port which the EJOE class loader server listens to
 	 * @param adapter
 	 *            the adapter used for (de)serializing input paramter objects for the server and the return values
 	 */
 	public EJClient(String host, int port, int classLoaderPort, final SerializeAdapter adapter)
 	{
 		this(host, port, adapter);
 		initClassLoader();
 	}
 
 	/***
 	 * Sets the connection timeout used when waiting for server responses. A value of zero (the default) blocks
 	 * indefinitely.
 	 *
 	 * @param timeout
 	 *            the new timeout in milliseconds
 	 */
 	public synchronized void setConnectionTimeout(int timeout)
 	{
 		this._connectionTimeout = timeout;
 	}
 
 	/***
 	 * Tells this client to use compression (if supported by the server) or not.
 	 *
 	 * @param enable
 	 */
 	public synchronized void enableCompression(boolean enable)
 	{
 		if (!this._requestInProgress)
 		{
 			this._clientInfo.setCompression(enable);
 		}
 		else
 		{
 			throw new IllegalStateException("Compression handling can't be changed while a request is in progress.");
 		}
 	}
 
 	/***
 	 * Tells this client to use compression (if supported by the server) with the given compression level.
 	 *
 	 * @param compressionLevel
 	 *            the level of compression to use, must be in range of 0-9
 	 */
 	public synchronized void enableCompression(int compressionLevel)
 	{
 		if (!this._requestInProgress)
 		{
 			this._clientInfo.setCompression(true);
 			this._clientInfo.setCompressionLevel(compressionLevel);
 		}
 		else
 		{
 			throw new IllegalStateException("Compression handling can't be changed while a request is in progress.");
 		}
 	}
 
 	/***
 	 * Enables/disables usage of a persistent connection. If persistent connection is disabled a new connection will be
 	 * used for each request.
 	 *
 	 * @param enable
 	 */
 	public synchronized void enablePersistentConnection(boolean enable)
 	{
 		if (!this._requestInProgress)
 		{
 			this._clientInfo.setPersistent(enable);
 		}
 		else
 		{
 			throw new IllegalStateException("Connection handling can't be changed while a request is in progress.");
 		}
 	}
 
 	/***
 	 * Enables remote classloading on the given remote port
 	 *
 	 * @param port
 	 *            port on which the EJOE classloader server listens
 	 */
 	public synchronized void enableRemoteClassloading(int port)
 	{
 		if (Thread.currentThread().getContextClassLoader() instanceof EJClassLoader)
 			throw new IllegalStateException("Remote classloading already enabled!");
 
 		this._classLoaderPort = port;
 		initClassLoader();
 	}
 
 	/***
 	 * Enables remote classloading on the default remote port.
 	 */
 	public synchronized void enableRemoteClassloading()
 	{
 		if (Thread.currentThread().getContextClassLoader() instanceof EJClassLoader)
 			throw new IllegalStateException("Remote classloading already enabled!");
 
 		initClassLoader();
 	}
 
 	/***
 	 * Main entry point for client tier implementations. Handles all remote server calls... This method is threadsafe,
 	 * ensuring that only one request is processed at a time.
 	 *
 	 * @param obj
 	 *            input objects for the EJOE Server
 	 * @return the object(s) returned by the EJOE server
 	 */
 	public synchronized Object execute(Object obj) throws IOException
 	{
 		Object result = null;
 
 		// ensure that only one request is in process at one time without
 		// blocking the whole client
 		this._requestInProgress = true;
 		ConnectionHeader serverInfo = null;
 		SocketChannel sChannel = null;
 		Selector selector = Selector.open();
 
 		try
 		{
 			sChannel = openSocketChannel();
 			sChannel.register(selector, SelectionKey.OP_CONNECT | SelectionKey.OP_WRITE);
 			Iterator it = null;
 			SelectionKey selKey = null;
 
 			while (selector.select(_connectionTimeout) > 0)
 			{
 				it = selector.selectedKeys().iterator();
 
 				while (it.hasNext())
 				{
 					selKey = (SelectionKey) it.next();
 					it.remove();
 
 					if (!selKey.isValid())
 					{
 						continue;
 					}
 
 					try
 					{
 						if (selKey.isConnectable() && sChannel.isConnectionPending())
 						{
 							if (!sChannel.finishConnect())
 							{
 								continue;
 							}
 
 							log.log(Level.INFO, "Connection established to "
 									+ sChannel.socket().getRemoteSocketAddress());
 						}
 						if (selKey.isWritable())
 						{
 							if (serverInfo == null)
 							{
 								log.log(Level.FINEST, "Handshaking server...");
 								serverInfo = IOUtils.handshake(sChannel, _clientInfo, true, _connectionTimeout);
 								if (serverInfo == null)
 								{
 									throw new IOException("Connection timeout (" + _connectionTimeout
 											+ "ms) reached while waiting for Handshake completing.");
 								}
 							}
 
 							if (serverInfo.hasNonBlockingReadWrite())
 							{
 								send(serverInfo, selKey, obj);
 								sChannel.register(selector, SelectionKey.OP_READ);
 							}
 							else
 							{
 								selKey.cancel();
 								sChannel.configureBlocking(true);
 								result = processBlocked(serverInfo, sChannel, obj);
 								return result;
 							}
 						}
 						if (selKey.isReadable())
 						{
 							result = receive(serverInfo, selKey);
 							return result;
 						}
 					}
 					catch (IncompleteIOException ioe)
 					{
 						selKey.cancel();
 						selector.selectNow();
 						sChannel.register(selector, ioe.getSelectionInterest(), ioe.getIOBuffer());
 					}
 					catch (IOException e)
 					{
 						selKey.cancel();
 						IOUtils.closeQuite(sChannel);
 						// rethrow origin exception to inform the caller
 						// without the hassle of nested exceptions
 						throw e;
 					}
 				}
 			}
 		}
 		finally
 		{
 			this._requestInProgress = false;
 			IOUtils.closeQuite(selector);
 
 			if (!this._clientInfo.isPersistent() || !serverInfo.isPersistent())
 			{
 				IOUtils.closeQuite(sChannel);
 			}
 		}
 
 		if (result instanceof Throwable)
 		{
 			throw new RemoteException("The server did return an Exception while handling your request!",
 					(Throwable) result);
 		}
 
 		return result;
 	}
 
 	/***
 	 * Closes an existing persistent connection to the corresponding EJOE server. Invoking this method when
 	 * non-persistent connections are used has no effect.
 	 */
 	public void close()
 	{
 		if (this._channel != null)
 		{
 			IOUtils.closeQuite(this._channel);
 			this._channel = null;
 		}
 	}
 
 	/***
 	 * Opens a new socket connection to the EJServer
 	 *
 	 * @return
 	 * @throws IOException
 	 */
 	private SocketChannel openSocketChannel() throws IOException
 	{
 		SocketChannel sChannel = this._channel;
 
 		if (sChannel == null || !sChannel.isOpen() || !this._clientInfo.isPersistent())
 		{
 			sChannel = SocketChannel.open();
 			sChannel.configureBlocking(false);
 			sChannel.connect(new InetSocketAddress(_host, _port));
 		}
 
 		return sChannel;
 	}
 
 	/***
 	 * Non-blocking request sending.
 	 *
 	 * @param serverInfo ConnectionHeader received from the EJServer
 	 * @param selKey
 	 * @param obj the request
 	 * @throws IOException
 	 */
 	private void send(ConnectionHeader serverInfo, SelectionKey selKey, Object obj) throws IOException
 	{
 		SocketChannel channel = (SocketChannel) selKey.channel();
 		ByteBuffer dataBuf = (ByteBuffer) selKey.attachment();
 		ByteBufferOutputStream out = null;
 
 		try
 		{
 			if (dataBuf == null)
 			{
 				out = new ByteBufferOutputStream();
 				serialize(out, obj, serverInfo);
 				dataBuf = out.getBackingBuffer();
 				dataBuf.flip();
 				log.log(Level.FINE, "Going to send request...");
 				IOUtils.writeHeader(serverInfo, dataBuf);
 			}
 
 			IOUtils.nonBlockingWrite(channel, dataBuf);
 			log.log(Level.FINE, "Request sent.");
 		}
 		finally
 		{
 			IOUtils.closeQuite(out);
 		}
 	}
 
 	/***
 	 * Non-blocking response reading.
 	 *
 	 * @param serverInfo ConnectionHeader received from the EJServer
 	 * @param selKey
 	 * @return server response received by the corresponding EJServer
 	 * @throws IOException
 	 */
 	private Object receive(ConnectionHeader serverInfo, SelectionKey selKey) throws IOException
 	{
 		Object result = null;
 		SocketChannel channel = (SocketChannel) selKey.channel();
 		ByteBuffer dataBuf = (ByteBuffer) selKey.attachment();
 		ByteBufferInputStream in = null;
 
 		try
 		{
 			if (dataBuf == null)
 			{
 				int length = IOUtils.readHeader(serverInfo);
 				// server signals a null result?
 				if (length == 0)
 				{
 					return null;
 				}
 				dataBuf = ByteBuffer.allocateDirect(length);
 				dataBuf.limit(length);
 				log.log(Level.FINE, "Going to read server response with length: " + length);
 			}
 
 			IOUtils.nonBlockingRead(channel, dataBuf);
 			dataBuf.flip();
 			log.log(Level.FINE, "Response read. Now calling (de)serialize adapter to convert response back to object.");
 
 			in = new ByteBufferInputStream(dataBuf);
 			result = deserialize(in, serverInfo);
 
 			log.log(Level.FINE, "Server response successfully converted to object.");
 
 			selKey.cancel();
 		}
 		finally
 		{
 			IOUtils.closeQuite(in);
 		}
 
 		return result;
 	}
 
 	/***
 	 * Combined blocking request sending and response reading.
 	 *
 	 * @param serverInfo
 	 *            ConnectionHeader received from the EJServer
 	 * @param channel
 	 *            the connected SocketChannel
 	 * @param obj
 	 *            the request
 	 * @return server response received by the corresponding EJServer
 	 * @throws IOException
 	 */
 	private Object processBlocked(ConnectionHeader serverInfo, SocketChannel channel, Object obj) throws IOException
 	{
 		Object result = null;
 
 		if (channel != null)
 		{
 			serialize(Channels.newOutputStream(channel), obj, serverInfo);
 			result = deserialize(Channels.newInputStream(channel), serverInfo);
 		}
 
 		return result;
 	}
 
 	/***
 	 * @param out
 	 * @param obj
 	 * @param serverInfo
 	 * @throws IOException
 	 */
 	private void serialize(OutputStream out, Object obj, ConnectionHeader serverInfo) throws IOException
 	{
 		boolean compressed = this._clientInfo.hasCompression() && serverInfo.hasCompression();
 		IOUtils.adapterSerialize(this._adapter, out, obj, compressed, this._clientInfo.getCompressionLevel());
 	}
 
 	/***
 	 * @param in
 	 * @param serverInfo
 	 * @return
 	 * @throws IOException
 	 */
 	private Object deserialize(InputStream in, ConnectionHeader serverInfo) throws IOException
 	{
 		boolean compressed = this._clientInfo.hasCompression() && serverInfo.hasCompression();
 		return IOUtils.adapterDeserialize(this._adapter, in, compressed);
 	}
 
 	/***
 	 * Reads the settings for this client from a given Properties instance
 	 *
 	 * @param props
 	 */
 	private void loadProperties(Properties props)
 	{
 		String clazz = null;
 
 		try
 		{
 			clazz = props.getProperty("ejoe.adapter", ObjectStreamAdapter.class.getName());
 			_adapter = (SerializeAdapter) Class.forName(clazz).newInstance();
 			this._clientInfo.setAdapterName(clazz);
 			_host = props.getProperty("ejoe.host", "127.0.0.1");
 			_port = Integer.parseInt(props.getProperty("ejoe.port", String.valueOf(EJConstants.EJOE_PORT)));
 			_connectionTimeout = Integer.parseInt(props.getProperty("ejoe.connectionTimeout", String
 					.valueOf(EJConstants.EJOE_CONNECTION_TIMEOUT)));
 			this._clientInfo.setCompression(Boolean.valueOf(props.getProperty("ejoe.compression", "false"))
 					.booleanValue());
 			this._clientInfo.setPersistent(Boolean.valueOf(props.getProperty("ejoe.persistentConnection", "true"))
 					.booleanValue());
 
 			if (Boolean.valueOf(props.getProperty("ejoe.remoteClassloader", "false")).booleanValue())
 			{
 				_classLoaderPort = Integer.parseInt(props.getProperty("ejoe.classLoaderPort", String
 						.valueOf(EJConstants.EJOE_CLASSLOADER_PORT)));
 				initClassLoader();
 			}
 		}
 		catch (InstantiationException e)
 		{
 			log.log(Level.SEVERE, "Can't instantiate class " + clazz);
 			throw new RuntimeException(e);
 		}
 		catch (IllegalAccessException e)
 		{
 			log.log(Level.SEVERE, "Can't access configured class " + clazz);
 			throw new RuntimeException(e);
 		}
 		catch (ClassNotFoundException e)
 		{
 			log.log(Level.SEVERE, "Can't locate configured class " + clazz);
 			throw new RuntimeException(e);
 		}
 	}
 
 	/***
 	 * Initializes a EJClassloader and sets it as the current context classloader. Also notifies the used
 	 * SerializeAdapter to recognize the classloader change.
 	 */
 	private void initClassLoader()
 	{
 		EJClassLoader ejcl = new EJClassLoader(Thread.currentThread().getContextClassLoader(), _host, _classLoaderPort);
 		Thread.currentThread().setContextClassLoader(ejcl);
 		this._adapter.handleClassLoaderChange(ejcl);
 	}
 }