package ca.sebastiendionne.grizzly.client;

import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.LinkedBlockingQueue;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import ca.sebastiendionne.grizzly.client.handler.ProtocolChainCallbackHandler;
import ca.sebastiendionne.grizzly.filter.EchoQueryManagerFilter;
import ca.sebastiendionne.grizzly.filter.EchoQueryProtocolFilter;
import ca.sebastiendionne.grizzly.manager.IEchoManager;
import ca.sebastiendionne.grizzly.queue.QueueConsumer;
import ca.sebastiendionne.grizzly.queue.QueueItem;
import ca.sebastiendionne.grizzly.server.ClientConnectionHandler;

import com.sun.grizzly.BaseSelectionKeyHandler;
import com.sun.grizzly.Controller;
import com.sun.grizzly.ControllerStateListenerAdapter;
import com.sun.grizzly.DefaultProtocolChain;
import com.sun.grizzly.DefaultProtocolChainInstanceHandler;
import com.sun.grizzly.ProtocolChain;
import com.sun.grizzly.ProtocolChainInstanceHandler;
import com.sun.grizzly.TCPConnectorHandler;
import com.sun.grizzly.TCPSelectorHandler;
import com.sun.grizzly.util.ByteBufferFactory;
import com.sun.grizzly.util.ConnectionCloseHandler;

public class GrizzlyClientProtocolChain implements IEchoManager {
	
	private static final Logger s_logger = LoggerFactory.getLogger(GrizzlyClientProtocolChain.class);
	
	private int port = 5000;
	private String host = "localhost";
	
	private Controller controller;	
	private TCPConnectorHandler connector_handler;
	
	//queue
	protected ThreadGroup f_tg = null;
	protected BlockingQueue<QueueItem> blockingQueue;
	protected QueueConsumer queueConsumer;
	
	
	private ByteBuffer buf = ByteBufferFactory.allocateView(1000, false);
	
	public void init() {
		
		f_tg = new ThreadGroup("Client Threads");
		
		final CountDownLatch started = new CountDownLatch(1);
	
		controller = new Controller();
		TCPSelectorHandler tcpSelectorHandler = new TCPSelectorHandler(true);
		controller.addSelectorHandler(tcpSelectorHandler);
	
		BaseSelectionKeyHandler selectionKeyHandler = new BaseSelectionKeyHandler();
		
		// to be notify when a client/server close the connection
		selectionKeyHandler.setConnectionCloseHandler(new ConnectionCloseHandler() {
	
			public void locallyClosed(SelectionKey key) {
					if(s_logger.isDebugEnabled()){
						s_logger.debug(key + " is being locally cancelled");
					}
			   }
	
			   public void remotlyClosed(SelectionKey key) {
				   if(s_logger.isDebugEnabled()){
					   s_logger.debug(key + " is being remotly cancelled (connection closed)");
				   }
			   }
		});
		
		tcpSelectorHandler.setSelectionKeyHandler(selectionKeyHandler);
		
		// STATE Listener
		controller.addStateListener(new ControllerStateListenerAdapter() {
	
			public void onException(Throwable e) {
				s_logger.error("Grizzly controller exception:" + e.getMessage());
			}
	
			public void onReady() {
				if(s_logger.isDebugEnabled()){
					s_logger.debug("Ready!");
				}
				started.countDown();
			}
	
		});
	
		
		// the protocol chain
		EchoQueryProtocolFilter protocolParser = new EchoQueryProtocolFilter();
		EchoQueryManagerFilter echoManagerFilter = new EchoQueryManagerFilter(this);
		   
		final ProtocolChain protocolChain = new DefaultProtocolChain();
		protocolChain.addFilter(protocolParser);
		protocolChain.addFilter(echoManagerFilter);
		((DefaultProtocolChain) protocolChain).setContinuousExecution(true);


		ProtocolChainInstanceHandler pciHandler = new DefaultProtocolChainInstanceHandler() {

           public boolean offer(ProtocolChain protocolChain) {
               return false;

           }

           public ProtocolChain poll() {

               return protocolChain;
           }
       };
       
       controller.setProtocolChainInstanceHandler(pciHandler);
		
	    // start the client
	    new Thread(f_tg, controller).start();
		
	    try {
			started.await();
		} catch (Exception e) {
			s_logger.error("Timeout in wait" + e.getMessage());
		}
	
		// extra stuff : Queue
		blockingQueue = new LinkedBlockingQueue<QueueItem>();
		queueConsumer = new QueueConsumer(this, blockingQueue);
		new Thread(f_tg, queueConsumer,"QueueConsumer").start();

		connector_handler = (TCPConnectorHandler) controller.acquireConnectorHandler(Controller.Protocol.TCP);
		
	}

	public void addToCache(SelectionKey key, ClientConnectionHandler ch){
		// useless in the client
	}

	public void removeFromCache(SelectionKey key){
		// useless in the client
	}
	
	public void addToQueue(QueueItem item){
		blockingQueue.add(item);
	}

	public void process(QueueItem item){
		s_logger.debug("Message from Server = " + item.getQuery());
	}
	
	
	public boolean connect() {
		try {
			connector_handler.connect(new InetSocketAddress(host, port), new ProtocolChainCallbackHandler(connector_handler));
		} catch (Exception e) {
			s_logger.error("Exception in execute..." + e);
			return false;
		}
		
		return true;
	}

	public void send(String quote) throws Exception {
		byte[] msg = quote.getBytes();
		
		if(msg.length>buf.capacity()){
			buf = ByteBufferFactory.allocateView(msg.length, false);
		}
		
		buf.put(msg);
		buf.flip();
		connector_handler.write(buf, true);
		buf.clear();
			
	}
	
	/**
	 * @param args
	 */
	public static void main(String[] args) {
		GrizzlyClientProtocolChain client = new GrizzlyClientProtocolChain();
		
		client.init();

		System.out.println("Initialization finished");

		client.connect();
		try {
			client.send("allo[EOQ]");
		} catch (Exception e) {
			e.printStackTrace();
		}

	}

}
