/*
    * Created on Nov 5, 2003
    * 
    */
   package org.neo.swarm.util.network.tcp;
   
   import java.net.InetSocketAddress;
   import java.net.ServerSocket;
   import java.nio.ByteBuffer;
  import java.nio.channels.SelectableChannel;
  import java.nio.channels.SelectionKey;
  import java.nio.channels.Selector;
  import java.nio.channels.ServerSocketChannel;
  import java.nio.channels.SocketChannel;
  import java.util.Iterator;
  import java.util.Set;
  import java.util.logging.Logger;
  
  import org.neo.swarm.util.threads.ThreadPool;
  
  /***
   * NBlocking ServerSocketFactory - pumps out incoming events onto the associated channels SocketStream 
   * that gets processed by the next available thread in the pool.
   * 
   * @author neil.avery
   */
  public class NIOTcpServer extends Thread implements TcpServiceAPI {
  
  	private static Logger logger = Logger.getLogger(NIOTcpServer.class.toString());	
  
  	private ThreadPool pool;
  	private boolean doListen = false;
  	private ListenCallback callback;
  	private java.net.InetAddress bind;
  	private int port;
  	private long timeout = 0;
  	private boolean isStarted = false;
  	private int bufferSize;
  
  	public NIOTcpServer(ListenCallback callback, ThreadPool pool, int bufferSize, java.net.InetAddress bind, int port, long timeout) {
  		this.pool = pool;
  		this.bufferSize = bufferSize;
  		this.callback = callback;
  		this.bind = bind;
  		this.port = port;
  		this.timeout = timeout;
  	}
  
  	public void run() {
  		try {
  			listen();
  		} catch (Exception x) {
  			logger.warning("Unable to start tcp server listener thread:"+ x + " bind:" + bind + " port:" + port);
  		}
  	}
  
  	public void listen() throws Exception {
  		doListen = true;
  		// allocate an unbound server socket channel
  		ServerSocketChannel serverChannel = ServerSocketChannel.open();
  		// Get the associated ServerSocket to bind it with
  		ServerSocket serverSocket = serverChannel.socket();
  		// create a new Selector for use below
  		Selector selector = Selector.open();
  		// set the port the server channel will listen to
  		serverSocket.bind(new InetSocketAddress(bind, port));
  		// set non-blocking mode for the listening socket
  		serverChannel.configureBlocking(false);
  		// register the ServerSocketChannel with the Selector
  		serverChannel.register(selector, SelectionKey.OP_ACCEPT);
  		logger.info("Starting tcp server on host:" + bind + " port:" + port);
  		setStarted(true);
  		while (doListen) {
  			// this may block for a long time, upon return the
  			// selected set contains keys of the ready channels
  			// workaround for bug in JDK1.4.2 nio libs...without this staggered sending cause staggereing
  			// reception of messages where the stagger == timeout period (recommended max 500ms).
  			int n = selector.select(500);
  			if (n == 0) {
  				n = selector.select(timeout);
  			}
  			if (n == 0) {
  				continue; // nothing to do
  			}
  
  			// get an iterator over the set of selected keys
  			Set keySet = selector.selectedKeys();
  			Iterator it = keySet.iterator();
  			// look at each key in the selected set
  			while (it.hasNext()) {
  				SelectionKey key = (SelectionKey) it.next();
  				// Is a new connection coming in?
  				if (key.isAcceptable()) {
  					ServerSocketChannel server = (ServerSocketChannel) key.channel();
  					SocketChannel channel = server.accept();
  					registerChannel(
  						selector,
  						channel,
  						SelectionKey.OP_READ,
 						new ObjectReader(channel, selector, callback, this.bufferSize));
 				}
 				// is there data to read on this channel?
 				if (key.isReadable()) {
 					readDataFromSocket(key);
 				}
 			} //while
 			keySet.clear();
 		} 
 		
 		serverChannel.close();
 		selector.close();
 		setStarted(false);
 		logger.info("Shutdown complete, tcp server on host:" + bind + " port:" + port);
 	}
 
 	public void stopListening() throws Exception {
 		doListen = false;
 		try {
 			Thread.sleep(timeout * 2);
 		} catch (InterruptedException e) {
 			e.printStackTrace();
 			logger.warning("stopListening() interrupted");
 			throw(e);
 		}
 	}
 
 	// ----------------------------------------------------------
 
 	/***
 	 * Register the given channel with the given selector for the given operations of interest
 	 */
 	protected void registerChannel(Selector selector, SelectableChannel channel, int ops, Object attach)
 		throws Exception {
 		if (channel == null)
 			return; // could happen
 		// set the new channel non-blocking
 		channel.configureBlocking(false);
 		// register it with the selector
 		channel.register(selector, ops, attach);
 	}
 
 	// ----------------------------------------------------------
 
 	// Use the same byte buffer for all channels. A single thread is
 	// servicing all the channels, so no danger of concurrent acccess.
 	private ByteBuffer buffer = ByteBuffer.allocateDirect(1024);
 
 	/***
 	 * Sample data handler method for a channel with data ready to read.
 	 * 
 	 * @param key
 	 *            A SelectionKey object associated with a channel determined by the selector to be ready for reading.
 	 *            If the channel returns an EOF condition, it is closed here, which automatically invalidates the
 	 *            associated key. The selector will then de-register the channel on the next select call.
 	 */
 	protected void readDataFromSocket(SelectionKey key) throws Exception {
 		ObjectReader reader = (ObjectReader) key.attachment();
 		// ask the ThreadPool to service this runnable
 		pool.execute(reader);
 	}
 	/***
 	 * @return Returns the isStarted.
 	 */
 	public boolean isStarted() {
 		return isStarted;
 	}
 
 	/***
 	 * @param isStarted
 	 *            The isStarted to set.
 	 */
 	private void setStarted(boolean isStarted) {
 		this.isStarted = isStarted;
 	}
 
 	/***
 	 * Start the server pool, processing threads and bind to the inet address
 	 */
 	public void startServer() {
 		this.start();
 		try {
 			while (this.isStarted() == false) {
 				Thread.sleep(10);
 			}
 		} catch (InterruptedException e) {
 			e.printStackTrace();
 			logger.warning("server startup interrupted:" + e);
 		}
 	}
 	/***
 	 * Stop the server from processing and more TCP incoming requests
 	 */
 	public void stopServer() throws Exception {
 		logger.info("about to stop tcp server");		
 		stopListening();
 		try {
 			this.pool.shutdown(1000);
 			// we can only wait for server exit when timeout is not set to 0
 			if (timeout == 0) {
 				logger.info("server 'stopServer' impossible due to TIMEOUT setting of '0'");
 			} else {
 				while (this.isStarted() == true) {
 					Thread.sleep(10);
 				}
 			}
 		} catch (InterruptedException e) {
 			e.printStackTrace();
 			logger.warning("server startup interrupted:" + e);
 		}
 	}
 
 }