package org.xvsm.remote.tcpconnection;
   
   import java.io.DataInputStream;
   import java.io.DataOutputStream;
   import java.io.EOFException;
   import java.io.IOException;
   import java.net.BindException;
   import java.net.InetAddress;
   import java.net.ServerSocket;
  import java.net.Socket;
  import java.net.SocketException;
  import java.net.URI;
  import java.net.URISyntaxException;
  import java.net.UnknownHostException;
  import java.util.HashMap;
  import java.util.concurrent.ExecutorService;
  import java.util.concurrent.Executors;
  
  import org.apache.log4j.Logger;
  import org.xvsm.internal.exceptions.FatalException;
  import org.xvsm.internal.tasks.Task;
  import org.xvsm.remote.TransportHandler;
  import org.xvsm.remote.interfaces.IMarshaller;
  import org.xvsm.remote.interfaces.ITransportListener;
  import org.xvsm.remote.interfaces.ITransportSender;
  
  public class TcpConnectedTransport implements ITransportListener<byte[]>,
  		ITransportSender<byte[]> {
  
  	private static Logger logger = Logger
  			.getLogger(TcpConnectedTransport.class);
  
  	/***
  	 * The maximal amount of Threads in the TcpListenerPool.
  	 */
  	private static final int MAX_THREADS = 10; // TODO: Get MAX_THREADS from
  
  	private HashMap<String, Socket> requestSockets = new HashMap<String, Socket>();
  
  	private HashMap<String, Socket> answerSockets = new HashMap<String, Socket>();
  	/***
  	 * The ServerSocket of the TcpTransportListener.
  	 */
  	private ServerSocket ssocket;
  
  	/***
  	 * Indicates if the TcpTransportListener is running or not.
  	 */
  	private boolean running = false;
  
  	/***
  	 * The Marshaller that should be used.
  	 */
  	private IMarshaller<byte[]> marshaller;
  
  	/***
  	 * The URI where the listener should listen.
  	 */
  	private URI uri;
  
  	private int socketPort;
  	/***
  	 * The ExecutorService to execute the TcpListenerThreads.
  	 */
  	private static ExecutorService service = Executors.newCachedThreadPool();
  
  	// newFixedThreadPool(MAX_THREADS);
  
  	/***
  	 * Default Constructor.
  	 */
  	public TcpConnectedTransport() {
  	}
  
  	/***
  	 * {@inheritDoc}
  	 */
  	public void run() {
  		Thread.currentThread().setName(this.getClass().getName());
  		this.running = true;
  		try {
  			ssocket = new ServerSocket(uri.getPort());
  			this.socketPort = ssocket.getLocalPort();
  			this.uri = new URI(this.uri.getScheme() + "://"
  					+ this.uri.getHost() + ":" + this.socketPort);
  			while (true) {
  				this.service.execute(new TcpListenerThread(ssocket.accept(),
  						true));
  			}
  		} catch (BindException e) {
  			throw new FatalException(e);
  		} catch (SocketException e) {
  			// Is thrown when the socket gets closed -> ignore
  		} catch (IOException e) {
  			throw new FatalException(e);
  		} catch (URISyntaxException e) {
  			throw new FatalException(e);
  		}
  	}
 
 	/***
 	 * {@inheritDoc}
 	 */
 	// @Override
 	public void stop() {
 		this.running = false;
 		try {
 			if (ssocket != null) {
 				this.ssocket.close();
 			}
 			for (Socket s : requestSockets.values()) {
 				s.close();
 			}
 			for (Socket s : answerSockets.values()) {
 				s.close();
 			}
 		} catch (IOException e) {
 			throw new FatalException(e);
 		}
 	}
 
 	/***
 	 * {@inheritDoc}
 	 */
 	// @Override
 	public void setMarshaller(IMarshaller<byte[]> m) {
 		this.marshaller = m;
 
 	}
 
 	public void setURI(URI nuri) {
 		this.uri = nuri;
 	}
 
 	/***
 	 * {@inheritDoc}
 	 */
 	// @Override
 	public URI getUri() {
 		return this.uri;
 	}
 
 	public synchronized void sendRequest(URI uri, Task t) {
 		try {
 			byte[] bytes = this.marshaller.marshall(t);
 			InetAddress ip;
 
 			if (InetAddress.getLocalHost().getHostAddress().equals(
 					uri.getHost())) {
 				ip = InetAddress.getByName("localhost");
 			} else {
 				ip = InetAddress.getByName(uri.getHost());
 			}
 			int port = uri.getPort();
 
 			Socket socket;
 			boolean createdSocket = false;
 
 			if ((socket = requestSockets.get(ip.getHostAddress() + ":" + port)) == null
 					|| socket.isClosed()) {
 				socket = new Socket(ip.getHostAddress(), port);
 				requestSockets.put(ip.getHostAddress() + ":" + port, socket);
 				createdSocket = true;
 			}
 
 			// Start a Listener on this socket
 			if (createdSocket) {
 				new TcpListenerThread(socket, false).start();
 			}
 
 			DataOutputStream out = new DataOutputStream(socket
 					.getOutputStream());
 			out.writeInt(bytes.length);
 			out.write(bytes);
 
 			out.flush();
 
 		} catch (UnknownHostException e) {
 			throw new FatalException(e);
 
 		} catch (IOException e) {
 			throw new FatalException("Exception during sending a request to "
 					+ uri, e);
 
 		}
 
 	}
 
 	public synchronized void sendResponse(URI uri, Task t) {
 		try {
 			byte[] bytes = this.marshaller.marshall(t);
 			InetAddress ip;
 
 			int port = uri.getPort();
 
 			Socket socket = null;
 			if ((socket = answerSockets.get(uri.toString())) == null
 					|| socket.isClosed()) {
 				if (InetAddress.getLocalHost().getHostAddress().equals(
 						uri.getHost())) {
 					ip = InetAddress.getByName("localhost");
 				} else {
 					ip = InetAddress.getByName(uri.getHost());
 				}
 				socket = new Socket(ip.getHostAddress(), port);
 				answerSockets.put(uri.toString(), socket);
 			}
 
 			DataOutputStream out = new DataOutputStream(socket
 					.getOutputStream());
 			out.writeInt(bytes.length);
 			out.write(bytes);
 			out.flush();
 		} catch (UnknownHostException e) {
 			throw new FatalException(e);
 
 		} catch (IOException e) {
 			throw new FatalException(e);
 
 		}
 	}
 
 	/***
 	 * 
 	 * @author Christian Schreiber, Michael Proestler
 	 * 
 	 */
 	class TcpListenerThread extends Thread {
 
 		/***
 		 * The Socket over which the communication takes place.
 		 */
 		private Socket socket;
 
 		private boolean newConnection;
 
 		private DataInputStream input;
 
 		private boolean listening;
 
 		/***
 		 * Default Constructor.
 		 * 
 		 * @param socket
 		 *            The socket over which the communication takes place.
 		 */
 		public TcpListenerThread(Socket socket, boolean newConnection) {
 			this.socket = socket;
 			this.listening = true;
 			this.newConnection = newConnection;
 			try {
 				this.input = new DataInputStream(this.socket.getInputStream());
 			} catch (IOException e) {
 				// TODO Auto-generated catch block
 				e.printStackTrace();
 			}
 		}
 
 		/***
 		 * {@inheritDoc}
 		 */
 		// @Override
 		public void run() {
 			Thread.currentThread().setName(this.getClass().getName());
 			try {
 				while (running && listening) {
 					try {
 
 						int dataSize = input.readInt();
 						int read;
 						int readTotal = 0;
 						byte[] data = new byte[dataSize];
 
 						while ((readTotal < dataSize)
 								&& (read = input.read(data, readTotal, dataSize
 										- readTotal)) != -1) {
 							readTotal += read;
 						}
 						Task task = marshaller.unmarshall(data);
 
 						if (newConnection) {
 							if (task.getAnswerToContainer() != null) {
 								String key = "";
 								int lastSlash = task.getAnswerToContainer()
 										.toString().lastIndexOf("/containers");
 								key = task.getAnswerToContainer().toString()
 										.substring(0, lastSlash);
 								answerSockets.put(key, this.socket);
 							}
 						}
 						TransportHandler.getInstance().processTask(task);
 					} catch (SocketException e) {
 						if (newConnection) {
 							logger.debug("Connection reset by peer");
 							this.listening = false;
 						} else {
 							logger.warn("Connection reset by peer");
 							this.listening = false;
 						}
 					}
 				}
 			} catch (EOFException e) {
 				logger
 						.debug("Connection lost: "
 								+ this.socket.getInetAddress());
 			} catch (IOException e) {
 				throw new FatalException(e);
 			} finally {
 				try {
 					this.socket.close();
 				} catch (IOException e) {
 					throw new FatalException(e);
 				}
 			}
 
 		}
 	}
 
 	/***
 	 * {@inheritDoc}
 	 */
 	@Override
 	public String toString() {
 		return "TcpConnectedTransport: MAX_THREADS:"
 				+ MAX_THREADS
 				+ " URI: "
 				+ this.uri
 				+ ((this.ssocket == null) ? "" : " port: "
 						+ this.ssocket.getLocalPort()) + " Marshaller: "
 				+ this.marshaller;
 	}
 }