// ========================================================================
// $Id: ThreadedServer.java,v 1.41 2005/12/10 00:38:20 gregwilkins Exp $
// Copyright 199-2004 Mort Bay Consulting Pty. Ltd.
// ------------------------------------------------------------------------
// 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.
// ========================================================================

package org.browsermob.proxy.jetty.util;

import org.apache.commons.logging.Log;
import org.browsermob.proxy.jetty.log.LogFactory;

import java.io.IOException;
import java.io.InputStream;
import java.io.InterruptedIOException;
import java.io.OutputStream;
import java.net.InetAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.UnknownHostException;

// TODO: Auto-generated Javadoc
/* ======================================================================= */
/**
 * Threaded socket server. This class listens at a socket and gives the
 * connections received to a pool of Threads
 * <P>
 * The class is abstract and derived classes must provide the handling for the
 * connections.
 * <P>
 * The properties THREADED_SERVER_MIN_THREADS and THREADED_SERVER_MAX_THREADS
 * can be set to control the number of threads created.
 * <P>
 * 
 * @version $Id: ThreadedServer.java,v 1.41 2005/12/10 00:38:20 gregwilkins Exp
 *          $
 * @author Greg Wilkins
 */
abstract public class ThreadedServer extends ThreadPool {

	/** The log. */
	private static Log log = LogFactory.getLog(ThreadedServer.class);

	/* ------------------------------------------------------------------- */
	/** The _address. */
	private InetAddrPort _address = null;

	/** The _so time out. */
	private int _soTimeOut = -1;

	/** The _linger time secs. */
	private int _lingerTimeSecs = 30;

	/** The _tcp no delay. */
	private boolean _tcpNoDelay = true;

	/** The _accept queue size. */
	private int _acceptQueueSize = 0;

	/** The _acceptors. */
	private int _acceptors = 1;

	/** The _acceptor. */
	private transient Acceptor[] _acceptor;

	/** The _listen. */
	private transient ServerSocket _listen = null;

	/** The _running. */
	private transient boolean _running = false;

	/* ------------------------------------------------------------------- */
	/*
	 * Construct
	 */
	/**
	 * Instantiates a new threaded server.
	 */
	public ThreadedServer() {
	}

	/* ------------------------------------------------------------ */
	/**
	 * Gets the server socket.
	 * 
	 * @return The ServerSocket
	 */
	public ServerSocket getServerSocket() {
		return _listen;
	}

	/* ------------------------------------------------------------------- */
	/**
	 * Construct for specific port.
	 * 
	 * @param port
	 *            the port
	 */
	public ThreadedServer(int port) {
		setInetAddrPort(new InetAddrPort(port));
	}

	/* ------------------------------------------------------------------- */
	/**
	 * Construct for specific address and port.
	 * 
	 * @param address
	 *            the address
	 * @param port
	 *            the port
	 */
	public ThreadedServer(InetAddress address, int port) {
		setInetAddrPort(new InetAddrPort(address, port));
	}

	/* ------------------------------------------------------------------- */
	/**
	 * Construct for specific address and port.
	 * 
	 * @param host
	 *            the host
	 * @param port
	 *            the port
	 * @throws UnknownHostException
	 *             the unknown host exception
	 */
	public ThreadedServer(String host, int port) throws UnknownHostException {
		setInetAddrPort(new InetAddrPort(host, port));
	}

	/* ------------------------------------------------------------------- */
	/**
	 * Construct for specific address and port.
	 * 
	 * @param address
	 *            the address
	 */
	public ThreadedServer(InetAddrPort address) {
		setInetAddrPort(address);
	}

	/* ------------------------------------------------------------ */
	/**
	 * Set the server InetAddress and port.
	 * 
	 * @param address
	 *            The Address to listen on, or 0.0.0.0:port for all interfaces.
	 */
	public synchronized void setInetAddrPort(InetAddrPort address) {
		if (_address != null && _address.equals(address))
			return;

		if (isStarted())
			log.warn(this + " is started");

		_address = address;
	}

	/* ------------------------------------------------------------ */
	/**
	 * Gets the inet addr port.
	 * 
	 * @return IP Address and port in a new Instance of InetAddrPort.
	 */
	public InetAddrPort getInetAddrPort() {
		if (_address == null)
			return null;
		return new InetAddrPort(_address);
	}

	/* ------------------------------------------------------------ */
	/**
	 * Sets the host.
	 * 
	 * @param host
	 *            the new host
	 * @throws UnknownHostException
	 *             the unknown host exception
	 */
	public synchronized void setHost(String host) throws UnknownHostException {
		if (_address != null && _address.getHost() != null
				&& _address.getHost().equals(host))
			return;

		if (isStarted())
			log.warn(this + " is started");

		if (_address == null)
			_address = new InetAddrPort(host, 0);
		else
			_address.setHost(host);
	}

	/* ------------------------------------------------------------ */
	/**
	 * Gets the host.
	 * 
	 * @return Host name
	 */
	public String getHost() {
		if (_address == null || _address.getInetAddress() == null)
			return null;
		return _address.getHost();
	}

	/* ------------------------------------------------------------ */
	/**
	 * Sets the inet address.
	 * 
	 * @param addr
	 *            the new inet address
	 */
	public synchronized void setInetAddress(InetAddress addr) {
		if (_address != null && _address.getInetAddress() != null
				&& _address.getInetAddress().equals(addr))
			return;

		if (isStarted())
			log.warn(this + " is started");

		if (_address == null)
			_address = new InetAddrPort(addr, 0);
		else
			_address.setInetAddress(addr);
	}

	/* ------------------------------------------------------------ */
	/**
	 * Gets the inet address.
	 * 
	 * @return IP Address
	 */
	public InetAddress getInetAddress() {
		if (_address == null)
			return null;
		return _address.getInetAddress();
	}

	/* ------------------------------------------------------------ */
	/**
	 * Sets the port.
	 * 
	 * @param port
	 *            the new port
	 */
	public synchronized void setPort(int port) {
		if (_address != null && _address.getPort() == port)
			return;

		if (isStarted())
			log.warn(this + " is started");

		if (_address == null)
			_address = new InetAddrPort(port);
		else
			_address.setPort(port);
	}

	/* ------------------------------------------------------------ */
	/**
	 * Gets the port.
	 * 
	 * @return port number
	 */
	public int getPort() {
		if (_address == null)
			return 0;
		return _address.getPort();
	}

	/* ------------------------------------------------------------ */
	/**
	 * Set Max Read Time.
	 * 
	 * @param ms
	 *            the new max read time ms
	 * @deprecated maxIdleTime is used instead.
	 */
	public void setMaxReadTimeMs(int ms) {
		log.warn("setMaxReadTimeMs is deprecated. Use setMaxIdleTimeMs()");
	}

	/* ------------------------------------------------------------ */
	/**
	 * Gets the max read time ms.
	 * 
	 * @return milliseconds
	 */
	public int getMaxReadTimeMs() {
		return getMaxIdleTimeMs();
	}

	/* ------------------------------------------------------------ */
	/**
	 * Sets the linger time secs.
	 * 
	 * @param ls
	 *            seconds to linger or -1 to disable linger.
	 */
	public void setLingerTimeSecs(int ls) {
		_lingerTimeSecs = ls;
	}

	/* ------------------------------------------------------------ */
	/**
	 * Gets the linger time secs.
	 * 
	 * @return seconds.
	 */
	public int getLingerTimeSecs() {
		return _lingerTimeSecs;
	}

	/* ------------------------------------------------------------ */
	/**
	 * Sets the tcp no delay.
	 * 
	 * @param tcpNoDelay
	 *            if true then setTcpNoDelay(true) is called on accepted
	 *            sockets.
	 */
	public void setTcpNoDelay(boolean tcpNoDelay) {
		_tcpNoDelay = tcpNoDelay;
	}

	/* ------------------------------------------------------------ */
	/**
	 * Gets the tcp no delay.
	 * 
	 * @return true if setTcpNoDelay(true) is called on accepted sockets.
	 */
	public boolean getTcpNoDelay() {
		return _tcpNoDelay;
	}

	/* ------------------------------------------------------------ */
	/**
	 * Gets the accept queue size.
	 * 
	 * @return Returns the acceptQueueSize or -1 if not set.
	 */
	public int getAcceptQueueSize() {
		return _acceptQueueSize;
	}

	/* ------------------------------------------------------------ */
	/**
	 * The size of the queue for unaccepted connections. If not set, will
	 * default to greater of maxThreads or 50.
	 * 
	 * @param acceptQueueSize
	 *            The acceptQueueSize to set.
	 */
	public void setAcceptQueueSize(int acceptQueueSize) {
		_acceptQueueSize = acceptQueueSize;
	}

	/* ------------------------------------------------------------ */
	/**
	 * Set the number of threads used to accept connections. This should
	 * normally be 1, except when multiple CPUs are available and low latency is
	 * a high priority.
	 * 
	 * @param n
	 *            the new acceptor threads
	 */
	public void setAcceptorThreads(int n) {
		_acceptors = n;
	}

	/* ------------------------------------------------------------ */
	/**
	 * Get the nmber of threads used to accept connections.
	 * 
	 * @return the acceptor threads
	 */
	public int getAcceptorThreads() {
		return _acceptors;
	}

	/* ------------------------------------------------------------------- */
	/**
	 * Handle new connection. This method should be overridden by the derived
	 * class to implement the required handling. It is called by a thread
	 * created for it and does not need to return until it has finished it's
	 * task
	 * 
	 * @param in
	 *            the in
	 * @param out
	 *            the out
	 */
	protected void handleConnection(InputStream in, OutputStream out) {
		throw new Error("Either handlerConnection must be overridden");
	}

	/* ------------------------------------------------------------------- */
	/**
	 * Handle new connection. If access is required to the actual socket,
	 * override this method instead of handleConnection(InputStream
	 * in,OutputStream out). The default implementation of this just calls
	 * handleConnection(InputStream in,OutputStream out).
	 * 
	 * @param connection
	 *            the connection
	 * @throws IOException
	 *             Signals that an I/O exception has occurred.
	 */
	protected void handleConnection(Socket connection) throws IOException {
		if (log.isDebugEnabled())
			log.debug("Handle " + connection);
		InputStream in = connection.getInputStream();
		OutputStream out = connection.getOutputStream();

		handleConnection(in, out);
		out.flush();

		in = null;
		out = null;
		connection.close();
	}

	/* ------------------------------------------------------------ */
	/**
	 * Handle Job. Implementation of ThreadPool.handle(), calls
	 * handleConnection.
	 * 
	 * @param job
	 *            A Connection.
	 */
	public void handle(Object job) {
		Socket socket = (Socket) job;
		try {
			if (_tcpNoDelay)
				socket.setTcpNoDelay(true);
			handleConnection(socket);
		} catch (Exception e) {
			log.debug("Connection problem", e);
		} finally {
			try {
				socket.close();
			} catch (Exception e) {
				log.debug("Connection problem", e);
			}
		}
	}

	/* ------------------------------------------------------------ */
	/**
	 * New server socket. Creates a new servers socket. May be overriden by
	 * derived class to create specialist serversockets (eg SSL).
	 * 
	 * @param address
	 *            Address and port
	 * @param acceptQueueSize
	 *            Accept queue size
	 * @return The new ServerSocket
	 * @throws IOException
	 *             Signals that an I/O exception has occurred.
	 */
	protected ServerSocket newServerSocket(InetAddrPort address,
			int acceptQueueSize) throws java.io.IOException {
		if (address == null)
			return new ServerSocket(0, acceptQueueSize);

		return new ServerSocket(address.getPort(), acceptQueueSize,
				address.getInetAddress());
	}

	/* ------------------------------------------------------------ */
	/**
	 * Accept socket connection. May be overriden by derived class to create
	 * specialist serversockets (eg SSL).
	 * 
	 * @param ignored
	 *            the ignored
	 * @param timeout
	 *            The time to wait for a connection. Normally passed the
	 *            ThreadPool maxIdleTime.
	 * @return Accepted Socket
	 * @deprecated use acceptSocket(int timeout)
	 */
	protected Socket acceptSocket(ServerSocket ignored, int timeout) {
		return acceptSocket(timeout);
	}

	/* ------------------------------------------------------------ */
	/**
	 * Accept socket connection. May be overriden by derived class to create
	 * specialist serversockets (eg SSL).
	 * 
	 * @param timeout
	 *            The time to wait for a connection. Normally passed the
	 *            ThreadPool maxIdleTime.
	 * @return Accepted Socket
	 */
	protected Socket acceptSocket(int timeout) {
		try {
			Socket s = null;

			if (_listen != null) {
				if (_soTimeOut != timeout) {
					_soTimeOut = timeout;
					_listen.setSoTimeout(_soTimeOut);
				}

				s = _listen.accept();

				try {
					if (getMaxIdleTimeMs() >= 0)
						s.setSoTimeout(getMaxIdleTimeMs());
					if (_lingerTimeSecs >= 0)
						s.setSoLinger(true, _lingerTimeSecs);
					else
						s.setSoLinger(false, 0);
				} catch (Exception e) {
					LogSupport.ignore(log, e);
				}
			}
			return s;
		} catch (java.net.SocketException e) {
			// TODO - this is caught and ignored due strange
			// exception from linux java1.2.v1a
			LogSupport.ignore(log, e);
		} catch (InterruptedIOException e) {
			LogSupport.ignore(log, e);
		} catch (IOException e) {
			log.warn(LogSupport.EXCEPTION, e);
		}
		return null;
	}

	/* ------------------------------------------------------------------- */
	/**
	 * Open the server socket. This method can be called to open the server
	 * socket in advance of starting the listener. This can be used to test if
	 * the port is available.
	 * 
	 * @throws IOException
	 *             if an error occurs
	 */
	public void open() throws IOException {
		if (_listen == null) {
			_listen = newServerSocket(_address, _acceptQueueSize);

			if (_address == null)
				_address = new InetAddrPort(_listen.getInetAddress(),
						_listen.getLocalPort());
			else {
				if (_address.getInetAddress() == null)
					_address.setInetAddress(_listen.getInetAddress());
				if (_address.getPort() == 0)
					_address.setPort(_listen.getLocalPort());
			}

			_soTimeOut = getMaxIdleTimeMs();
			if (_soTimeOut >= 0)
				_listen.setSoTimeout(_soTimeOut);
		}
	}

	/* ------------------------------------------------------------------- */
	/*
	 * Start the ThreadedServer listening
	 */
	/*
	 * (non-Javadoc)
	 * 
	 * @see org.browsermob.proxy.jetty.util.ThreadPool#start()
	 */
	public synchronized void start() throws Exception {
		try {
			if (isStarted())
				return;

			open();

			_running = true;
			_acceptor = new Acceptor[_acceptors];
			for (int a = 0; a < _acceptor.length; a++) {
				_acceptor[a] = new Acceptor();
				_acceptor[a].setDaemon(isDaemon());
				_acceptor[a].start();
			}

			super.start();
		} catch (Exception e) {
			log.warn("Failed to start: " + this);
			throw e;
		}
	}

	/* --------------------------------------------------------------- */
	/*
	 * (non-Javadoc)
	 * 
	 * @see org.browsermob.proxy.jetty.util.ThreadPool#stop()
	 */
	public void stop() throws InterruptedException {
		synchronized (this) {
			// Signal that we are stopping
			_running = false;

			// Close the listener socket.
			if (log.isDebugEnabled())
				log.debug("closing " + _listen);
			try {
				if (_listen != null)
					_listen.close();
				_listen = null;
			} catch (IOException e) {
				log.warn(LogSupport.EXCEPTION, e);
			}

			// Do we have an acceptor thread (running or not)
			Thread.yield();
			for (int a = 0; _acceptor != null && a < _acceptor.length; a++) {
				Acceptor acc = _acceptor[a];
				if (acc != null)
					acc.interrupt();
			}
			Thread.sleep(100);

			for (int a = 0; _acceptor != null && a < _acceptor.length; a++) {
				Acceptor acc = _acceptor[a];

				if (acc != null) {
					acc.forceStop();
					_acceptor[a] = null;
				}
			}
		}

		// Stop the thread pool
		try {
			super.stop();
		} catch (Exception e) {
			log.warn(LogSupport.EXCEPTION, e);
		} finally {
			synchronized (this) {
				_acceptor = null;
			}
		}
	}

	/* ------------------------------------------------------------ */
	/* ------------------------------------------------------------ */
	/**
	 * Kill a job. This method closes IDLE and socket associated with a job
	 * 
	 * @param thread
	 *            the thread
	 * @param job
	 *            the job
	 */
	protected void stopJob(Thread thread, Object job) {
		if (job instanceof Socket) {
			try {
				((Socket) job).close();
			} catch (Exception e) {
				LogSupport.ignore(log, e);
			}
		}
		super.stopJob(thread, job);
	}

	/* ------------------------------------------------------------ */
	/*
	 * (non-Javadoc)
	 * 
	 * @see java.lang.Object#toString()
	 */
	public String toString() {
		if (_address == null)
			return getName() + "@0.0.0.0:0";
		if (_listen != null)
			return getName() + "@" + _listen.getInetAddress().getHostAddress()
					+ ":" + _listen.getLocalPort();
		return getName() + "@" + getInetAddrPort();
	}

	/* ------------------------------------------------------------ */
	/* ------------------------------------------------------------ */
	/* ------------------------------------------------------------ */
	/**
	 * The Class Acceptor.
	 */
	private class Acceptor extends Thread {
		/* ------------------------------------------------------------ */
		/*
		 * (non-Javadoc)
		 * 
		 * @see java.lang.Thread#run()
		 */
		public void run() {
			ThreadedServer threadedServer = ThreadedServer.this;
			try {
				this.setName("Acceptor " + _listen);
				while (_running) {
					try {
						// Accept a socket
						Socket socket = acceptSocket(_soTimeOut);

						// Handle the socket
						if (socket != null) {
							if (_running)
								threadedServer.run(socket);
							else
								socket.close();
						}
					} catch (Throwable e) {
						if (_running)
							log.warn(LogSupport.EXCEPTION, e);
						else
							log.debug(LogSupport.EXCEPTION, e);
					}
				}
			} finally {
				if (_running)
					log.warn("Stopping " + this.getName());
				else
					log.info("Stopping " + this.getName());
				synchronized (threadedServer) {
					if (_acceptor != null) {
						for (int a = 0; a < _acceptor.length; a++)
							if (_acceptor[a] == this)
								_acceptor[a] = null;
					}
					threadedServer.notifyAll();
				}
			}
		}

		/* ------------------------------------------------------------ */
		/**
		 * Force stop.
		 */
		void forceStop() {
			if (_listen != null && _address != null) {
				InetAddress addr = _address.getInetAddress();
				try {
					if (addr == null || addr.toString().startsWith("0.0.0.0"))
						addr = InetAddress.getByName("127.0.0.1");
					if (log.isDebugEnabled())
						log.debug("Self connect to close listener " + addr
								+ ":" + _address.getPort());
					Socket socket = new Socket(addr, _address.getPort());
					Thread.yield();
					socket.close();
					Thread.yield();
				} catch (IOException e) {
					if (log.isDebugEnabled())
						log.debug("problem stopping acceptor " + addr + ": ", e);
				}
			}
		}
	}

}
