//
//  ========================================================================
//  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.nio.channels.ServerSocketChannel;
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 final int _maxIdleTime = 200000;
	protected int _soLingerTime = -1;

	protected final HttpBuffersImpl _buffers = new HttpBuffersImpl();

	// from child classes
	protected transient ServerSocketChannel _acceptChannel;

	public Connector(Server server,int port) {
		this.server = server;
		this.port = port;
		server.connectors.add(this);
		addBean(_buffers);
		_buffers.setRequestBufferType(Type.DIRECT);
		_buffers.setRequestHeaderType(Type.INDIRECT);
		_buffers.setResponseBufferType(Type.DIRECT);
		_buffers.setResponseHeaderType(Type.INDIRECT);
	}

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

	public String getHost()
	{
		return _host;
	}

	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 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
	{
		super.doStart();

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

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

	@Override
	protected synchronized void doStop() throws Exception
	{
		try
		{
			if (_acceptChannel != null)
				_acceptChannel.close();
			_acceptChannel=null;
		}
		catch (IOException e)
		{
			LOG.warn("",e);
		}

		super.doStop();
	}


	protected final void configure(Socket socket) throws IOException
	{
		socket.setTcpNoDelay(true);
		if (_soLingerTime >= 0)
			socket.setSoLinger(true,_soLingerTime / 1000);
		else
			socket.setSoLinger(false,0);
	}

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

	public void persist(EndPoint endpoint) throws IOException
	{
	}

	public boolean isConfidential(Request request)
	{
		return false;
	}

	protected abstract void accept() 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(),
				port);
	}


	private class Acceptor implements Runnable
	{

		public void run()
		{
			Thread current = Thread.currentThread();
			String name = current.getName();
			current.setName(name + " Acceptor" + " " + Connector.this);

			try
			{
				while (isRunning() && _acceptChannel != null)
				{
					try
					{
						accept();
					}
					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);
			}
		}
	}

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

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


	// from AbstractNIOConnector

	/* ------------------------------------------------------------------------------- */
	public boolean getUseDirectBuffers()
	{
		return getRequestBufferType()==Type.DIRECT;
	}

	/* ------------------------------------------------------------------------------- */
	/**
	 * @param direct If True (the default), the connector can use NIO direct buffers.
	 * Some JVMs have memory management issues (bugs) with direct buffers.
	 */
	public void setUseDirectBuffers(boolean direct)
	{
		_buffers.setRequestBufferType(direct?Type.DIRECT:Type.INDIRECT);
		_buffers.setResponseBufferType(direct?Type.DIRECT:Type.INDIRECT);
	}


	// from child classes

}