//
//  ========================================================================
//  Copyright (c) 1995-2014 Mort Bay Consulting Pty. Ltd.
//  ------------------------------------------------------------------------
//  All rights reserved. This program and the accompanying materials
//  are made available under the terms of the Eclipse Public License v1.0
//  and Apache License v2.0 which accompanies this distribution.
//
//      The Eclipse Public License is available at
//      http://www.eclipse.org/legal/epl-v10.html
//
//      The Apache License v2.0 is available at
//      http://www.opensource.org/licenses/apache2.0.php
//
//  You may elect to redistribute this code under either of these licenses.
//  ========================================================================
//

package org.eclipse.jetty.server;

import java.io.IOException;
import java.net.InetAddress;
import java.net.Socket;
import java.net.UnknownHostException;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.atomic.AtomicLong;

import javax.servlet.ServletRequest;

import org.eclipse.jetty.http.HttpBuffers;
import org.eclipse.jetty.http.HttpBuffersImpl;
import org.eclipse.jetty.http.HttpFields;
import org.eclipse.jetty.http.HttpHeaders;
import org.eclipse.jetty.http.HttpSchemes;
import org.eclipse.jetty.io.Buffers;
import org.eclipse.jetty.io.Buffers.Type;
import org.eclipse.jetty.io.Connection;
import org.eclipse.jetty.io.EndPoint;
import org.eclipse.jetty.io.EofException;
import org.eclipse.jetty.util.component.AggregateLifeCycle;
import org.eclipse.jetty.util.component.Dumpable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Abstract Connector implementation. This abstract implementation of the Connector interface provides:
 * <ul>
 * <li>AbstractLifeCycle implementation</li>
 * <li>Implementations for connector getters and setters</li>
 * <li>Buffer management</li>
 * <li>Socket configuration</li>
 * <li>Base acceptor thread</li>
 * <li>Optional reverse proxy headers checking</li>
 * </ul>
 */
public abstract class Connector extends AggregateLifeCycle implements HttpBuffers, Dumpable
{
	private static final Logger LOG = LoggerFactory.getLogger(Connector.class);

	private String _name;

	public final Server server;
	private String _host;
	public final int port;
	private int _acceptors = 1;

	protected int _maxIdleTime = 200000;
	protected int _lowResourceMaxIdleTime = -1;
	protected int _soLingerTime = -1;

	private transient Thread[] _acceptorThreads;

	protected final HttpBuffersImpl _buffers = new HttpBuffersImpl();

	public Connector(Server server,int port) {
		this.server = server;
		this.port = port;
		server.connectors.add(this);
		addBean(_buffers);
	}

	/* ------------------------------------------------------------ */
	/**
	 */
	public void setHost(String host)
	{
		_host = host;
	}

	/* ------------------------------------------------------------ */
	/*
	 */
	public String getHost()
	{
		return _host;
	}

	/* ------------------------------------------------------------ */
	/**
	 * @return Returns the maxIdleTime.
	 */
	public int getMaxIdleTime()
	{
		return _maxIdleTime;
	}

	/* ------------------------------------------------------------ */
	/**
	 * Set the maximum Idle time for a connection, which roughly translates to the {@link Socket#setSoTimeout(int)} call, although with NIO implementations
	 * other mechanisms may be used to implement the timeout. The max idle time is applied:
	 * <ul>
	 * <li>When waiting for a new request to be received on a connection</li>
	 * <li>When reading the headers and content of a request</li>
	 * <li>When writing the headers and content of a response</li>
	 * </ul>
	 * Jetty interprets this value as the maximum time between some progress being made on the connection. So if a single byte is read or written, then the
	 * timeout (if implemented by jetty) is reset. However, in many instances, the reading/writing is delegated to the JVM, and the semantic is more strictly
	 * enforced as the maximum time a single read/write operation can take. Note, that as Jetty supports writes of memory mapped file buffers, then a write may
	 * take many 10s of seconds for large content written to a slow device.
	 * <p>
	 * Previously, Jetty supported separate idle timeouts and IO operation timeouts, however the expense of changing the value of soTimeout was significant, so
	 * these timeouts were merged. With the advent of NIO, it may be possible to again differentiate these values (if there is demand).
	 *
	 * @param maxIdleTime
	 *            The maxIdleTime to set.
	 */
	public void setMaxIdleTime(int maxIdleTime)
	{
		_maxIdleTime = maxIdleTime;
	}

	/* ------------------------------------------------------------ */
	/**
	 * @return Returns the maxIdleTime when resources are low.
	 */
	public int getLowResourcesMaxIdleTime()
	{
		return _lowResourceMaxIdleTime;
	}

	/* ------------------------------------------------------------ */
	/**
	 * @param maxIdleTime
	 *            The maxIdleTime to set when resources are low.
	 */
	public void setLowResourcesMaxIdleTime(int maxIdleTime)
	{
		_lowResourceMaxIdleTime = maxIdleTime;
	}

	/* ------------------------------------------------------------ */
	/**
	 * @return Returns the maxIdleTime when resources are low.
	 * @deprecated
	 */
	@Deprecated
	public final int getLowResourceMaxIdleTime()
	{
		return getLowResourcesMaxIdleTime();
	}

	/* ------------------------------------------------------------ */
	/**
	 * @param maxIdleTime
	 *            The maxIdleTime to set when resources are low.
	 * @deprecated
	 */
	@Deprecated
	public final void setLowResourceMaxIdleTime(int maxIdleTime)
	{
		setLowResourcesMaxIdleTime(maxIdleTime);
	}

	/* ------------------------------------------------------------ */
	/**
	 * @return Returns the soLingerTime.
	 */
	public int getSoLingerTime()
	{
		return _soLingerTime;
	}

	/* ------------------------------------------------------------ */
	/**
	 * @return Returns the number of acceptor threads.
	 */
	public int getAcceptors()
	{
		return _acceptors;
	}

	/* ------------------------------------------------------------ */
	/**
	 * @param acceptors
	 *            The number of acceptor threads to set.
	 */
	public void setAcceptors(int acceptors)
	{
		if (acceptors > 2 * Runtime.getRuntime().availableProcessors())
			LOG.warn("Acceptors should be <=2*availableProcessors: " + this);
		_acceptors = acceptors;
	}

	/* ------------------------------------------------------------ */
	/**
	 * @param soLingerTime
	 *            The soLingerTime to set or -1 to disable.
	 */
	public void setSoLingerTime(int soLingerTime)
	{
		_soLingerTime = soLingerTime;
	}

	/* ------------------------------------------------------------ */
	@Override
	protected void doStart() throws Exception
	{
		// open listener port
		open();

		super.doStart();

		// Start selector thread
		synchronized (this)
		{
			_acceptorThreads = new Thread[getAcceptors()];

			ThreadPoolExecutor _threadPool = server.threadPool;
			for (int i = 0; i < _acceptorThreads.length; i++)
				_threadPool.execute(new Acceptor(i));
			if (server.isLowOnThreads())
				LOG.warn("insufficient threads configured for {}",this);
		}

		LOG.info("Started {}",this);
	}

	/* ------------------------------------------------------------ */
	@Override
	protected void doStop() throws Exception
	{
		try
		{
			close();
		}
		catch (IOException e)
		{
			LOG.warn("",e);
		}

		super.doStop();

		Thread[] acceptors;
		synchronized (this)
		{
			acceptors = _acceptorThreads;
			_acceptorThreads = null;
		}
		if (acceptors != null)
		{
			for (Thread thread : acceptors)
			{
				if (thread != null)
					thread.interrupt();
			}
		}
	}

	/* ------------------------------------------------------------ */
	public void join() throws InterruptedException
	{
		Thread[] threads;
		synchronized(this)
		{
			threads=_acceptorThreads;
		}
		if (threads != null)
			for (Thread thread : threads)
				if (thread != null)
					thread.join();
	}

	/* ------------------------------------------------------------ */
	protected void configure(Socket socket) throws IOException
	{
		try
		{
			socket.setTcpNoDelay(true);
			if (_soLingerTime >= 0)
				socket.setSoLinger(true,_soLingerTime / 1000);
			else
				socket.setSoLinger(false,0);
		}
		catch (Exception e)
		{
			LOG.trace("",e);
		}
	}

	public void customize(EndPoint endpoint, Request request) throws IOException
	{
	}

	public void persist(EndPoint endpoint) throws IOException
	{
	}

	/* ------------------------------------------------------------ */
	/*
	 * @see org.eclipse.jetty.server.Connector#isConfidential(org.eclipse.jetty.server.Request)
	 */
	public boolean isConfidential(Request request)
	{
		return false;
	}

	/* ------------------------------------------------------------ */
	protected abstract void accept(int acceptorID) throws IOException, InterruptedException;

	/* ------------------------------------------------------------ */
	public void stopAccept(int acceptorID) throws Exception
	{
	}

	public int getRequestBufferSize()
	{
		return _buffers.getRequestBufferSize();
	}

	public void setRequestBufferSize(int requestBufferSize)
	{
		_buffers.setRequestBufferSize(requestBufferSize);
	}

	public int getRequestHeaderSize()
	{
		return _buffers.getRequestHeaderSize();
	}

	public void setRequestHeaderSize(int requestHeaderSize)
	{
		_buffers.setRequestHeaderSize(requestHeaderSize);
	}

	public int getResponseBufferSize()
	{
		return _buffers.getResponseBufferSize();
	}

	public void setResponseBufferSize(int responseBufferSize)
	{
		_buffers.setResponseBufferSize(responseBufferSize);
	}

	public int getResponseHeaderSize()
	{
		return _buffers.getResponseHeaderSize();
	}

	public void setResponseHeaderSize(int responseHeaderSize)
	{
		_buffers.setResponseHeaderSize(responseHeaderSize);
	}

	public Type getRequestBufferType()
	{
		return _buffers.getRequestBufferType();
	}

	public Type getRequestHeaderType()
	{
		return _buffers.getRequestHeaderType();
	}

	public Type getResponseBufferType()
	{
		return _buffers.getResponseBufferType();
	}

	public Type getResponseHeaderType()
	{
		return _buffers.getResponseHeaderType();
	}

	public void setRequestBuffers(Buffers requestBuffers)
	{
		_buffers.setRequestBuffers(requestBuffers);
	}

	public void setResponseBuffers(Buffers responseBuffers)
	{
		_buffers.setResponseBuffers(responseBuffers);
	}

	public Buffers getRequestBuffers()
	{
		return _buffers.getRequestBuffers();
	}

	public Buffers getResponseBuffers()
	{
		return _buffers.getResponseBuffers();
	}

	public int getMaxBuffers()
	{
		return _buffers.getMaxBuffers();
	}

	/* ------------------------------------------------------------ */
	@Override
	public String toString()
	{
		return String.format("%s@%s:%d",
				getClass().getSimpleName(),
				getHost()==null?"0.0.0.0":getHost(),
				getLocalPort()<=0 ? port : getLocalPort());
	}

	/* ------------------------------------------------------------ */
	/* ------------------------------------------------------------ */
	/* ------------------------------------------------------------ */
	private class Acceptor implements Runnable
	{
		int _acceptor = 0;

		Acceptor(int id)
		{
			_acceptor = id;
		}

		/* ------------------------------------------------------------ */
		public void run()
		{
			Thread current = Thread.currentThread();
			String name;
			synchronized (Connector.this)
			{
				if (_acceptorThreads == null)
					return;

				_acceptorThreads[_acceptor] = current;
				name = _acceptorThreads[_acceptor].getName();
				current.setName(name + " Acceptor" + _acceptor + " " + Connector.this);
			}

			try
			{
				while (isRunning() && getConnection() != null)
				{
					try
					{
						accept(_acceptor);
					}
					catch (EofException e)
					{
						LOG.trace("",e);
					}
					catch (IOException e)
					{
						LOG.trace("",e);
					}
					catch (InterruptedException x)
					{
						// Connector has been stopped
						LOG.trace("",x);
					}
					catch (Throwable e)
					{
						LOG.warn("",e);
					}
				}
			}
			finally
			{
				current.setName(name);

				synchronized (Connector.this)
				{
					if (_acceptorThreads != null)
						_acceptorThreads[_acceptor] = null;
				}
			}
		}
	}

	public String getName()
	{
		if (_name == null)
			_name = (getHost() == null?"0.0.0.0":getHost()) + ":" + (getLocalPort() <= 0?port:getLocalPort());
		return _name;
	}

	/* ------------------------------------------------------------ */
	public final boolean isLowResources()
	{
		return server.isLowOnThreads();
	}

	// from old interface
	public abstract void open() throws IOException;
	public abstract void close() throws IOException;
	public abstract int getLocalPort();
	public abstract Object getConnection();
}