/***
    * 
    */
   package org.xvsm.internal;
   
   import java.net.URI;
   import java.util.ArrayList;
   import java.util.List;
   import java.util.Properties;
  import java.util.concurrent.locks.Lock;
  import java.util.concurrent.locks.ReadWriteLock;
  import java.util.concurrent.locks.ReentrantReadWriteLock;
  
  import org.apache.log4j.Logger;
  import org.xvsm.core.ContainerRef;
  import org.xvsm.core.Entry;
  import org.xvsm.core.VoidEntry;
  import org.xvsm.core.aspect.IAspect;
  import org.xvsm.core.aspect.IPoint;
  import org.xvsm.interfaces.ICapi;
  import org.xvsm.interfaces.ICoordinator;
  import org.xvsm.interfaces.container.IContainer;
  import org.xvsm.interfaces.container.ITransactionLayer;
  import org.xvsm.internal.exceptions.AspectNotOkException;
  import org.xvsm.internal.exceptions.AspectRescheduleException;
  import org.xvsm.internal.exceptions.ContainerFullException;
  import org.xvsm.internal.exceptions.CountNotMetException;
  import org.xvsm.internal.exceptions.FatalException;
  import org.xvsm.internal.exceptions.InvalidContainerException;
  import org.xvsm.internal.exceptions.InvalidTransactionException;
  import org.xvsm.internal.exceptions.TimeoutExpiredException;
  import org.xvsm.internal.exceptions.TransactionLockException;
  import org.xvsm.internal.tasks.OperationTask;
  import org.xvsm.internal.tasks.dao.IOperationTaskDAO;
  import org.xvsm.internal.tasks.dao.ITransactionDAO;
  import org.xvsm.internal.tasks.dao.inmemory.OperationTaskDAO;
  import org.xvsm.internal.tasks.dao.inmemory.TransactionDAO;
  import org.xvsm.selectors.Selector;
  import org.xvsm.transactions.Transaction;
  
  /***
   * @author Christian Schreiber, Michael Proestler
   * 
   */
  public class BlockingLayer implements IContainer {
  
  	/***
  	 * the logger.
  	 */
  	private static Logger logger = Logger.getLogger(BlockingLayer.class);
  
  	/***
  	 * The transaction layer for this container.
  	 */
  	private ITransactionLayer c;
  
  	/***
  	 * The lock.
  	 */
  	private ReadWriteLock rwLock = new ReentrantReadWriteLock();
  
  	/***
  	 * Contains all tasks which block because another transaction uses the
  	 * container.
  	 */
  	private ITransactionDAO transactionDAO = new TransactionDAO();
  
  	/***
  	 * All blocking operations.
  	 * 
  	 */
  	// TODO make separate DAOs for read/write/....?
  	private IOperationTaskDAO blockingOperations = new OperationTaskDAO();
  
  	/***
  	 * Creates a new BlockingLayer.
  	 * 
  	 * @param size
  	 *            the size of the new container.
  	 */
  	public BlockingLayer(int size) {
  		c = new TransactionLayer(size);
  	}
  
  	/***
  	 * {@inheritDoc}.
  	 */
  	public Object execute(OperationTask task) throws ContainerFullException,
  			CountNotMetException, TransactionLockException,
  			AspectNotOkException, AspectRescheduleException {
  		if (logger.isDebugEnabled()) {
  			logger.debug("execute() " + task + " -- " + this.getCref());
  		}
  		Lock lock = null;
  		Object result = null;
  		// this.transactionDAO.writeLock().lock();
  		try {
  			switch (task.getType()) {
  			case READ:
 				lock = rwLock.readLock();
 				lock.lock();
 				result = c.read(task.getTx(), task.getSelectors(), task
 						.getRetrycount(), task.getAspectContext());
 				break;
 			case WRITE:
 				lock = rwLock.writeLock();
 				lock.lock();
 				c.write(task.getEntries(), task.getTx(), task.getRetrycount(),
 						task.getAspectContext());
 				result = new VoidEntry();
 				break;
 			case TAKE:
 				lock = rwLock.writeLock();
 				lock.lock();
 				result = c.take(false, task.getTx(), task.getSelectors(), task
 						.getRetrycount(), task.getAspectContext());
 				break;
 			case SHIFT:
 
 				lock = rwLock.writeLock();
 				lock.lock();
 				c.shift(task.getEntries(), task.getTx(), task
 						.getAspectContext());
 				result = new VoidEntry();
 				break;
 			case DESTROY:
 
 				lock = rwLock.writeLock();
 				lock.lock();
 				List<Entry> deleted = c.take(true, task.getTx(), task
 						.getSelectors(), task.getRetrycount(), task
 						.getAspectContext());
 				task.setDeleted(deleted);
 				result = new VoidEntry();
 				break;
 			default:
 				break;
 			}
 			// increase retrycount.
 			task.setRetrycount(task.getRetrycount() + 1);
 
 			this.wakeUpBlocking();
 			if (logger.isDebugEnabled()) {
 				logger.debug("execute() executed task result: " + result
 						+ " -- " + this.getCref());
 			}
 			return result;
 		} catch (ContainerFullException e) {
 			if (logger.isDebugEnabled()) {
 				logger.debug("execute() NoPlaceLeftException was thrown."
 						+ " Adding task to blocking operations -- "
 						+ this.getCref());
 			}
 			this.blockOperationTask(task, e);
 			throw e;
 		} catch (CountNotMetException e) {
 			if (logger.isDebugEnabled()) {
 				logger.debug("execute() CountNotMetException was thrown."
 						+ " Adding task to blocking operations -- "
 						+ this.getCref());
 			}
 			this.blockOperationTask(task, e);
 			throw e;
 		} catch (TransactionLockException e) {
 			if (!this.transactionDAO.add(task, e.getReason())) {
 				// A transactions commited in the time
 				EventProcessingPool.getInstance().execute(task);
 			}
 			throw e;
 		} catch (InvalidTransactionException e) {
 			return e;
 		} catch (AspectRescheduleException e) {
 			// increase retrycount.
 			task.setRetrycount(task.getRetrycount() + 1);
 			throw e;
 		} catch (Exception e) {
 			return new FatalException(e);
 		} finally {
 			// release the lock.
 			if (lock != null) {
 				lock.unlock();
 				lock = null;
 			}
 		}
 	}
 
 	/***
 	 * Adds the Task to the blocking Tasks if the timeout is INFINITE or > 0.
 	 * Otherwise the result of the task is set to e.
 	 * 
 	 * @param task
 	 *            the task to block.
 	 * @param e
 	 *            the exception that will be set as result, if the timeout
 	 *            expired.
 	 */
 	private void blockOperationTask(OperationTask task, Exception e) {
 		// Block the task if the timeout is > 0 or INFINITE. Otherwise
 		// return the exception.
 		if (task.getTimeout() == ICapi.INFINITE_TIMEOUT
 				|| task.getTimeout() > 0) {
 			blockingOperations.add(task);
 			if (logger.isDebugEnabled()) {
 				logger.debug("blockOperationTask() added task");
 			}
 		} else {
 			if (logger.isDebugEnabled()) {
 				logger.debug("blockOperationTask() timeout is 0 => "
 						+ "set the exception " + e + " as result");
 			}
 			task.setResult(e);
 		}
 	}
 
 	/***
 	 * 
 	 * {@inheritDoc}.
 	 */
 	public ContainerRef getCref() {
 		return c.getCref();
 	}
 
 	/***
 	 * 
 	 * {@inheritDoc}.
 	 */
 	public void setCref(ContainerRef cref) {
 		c.setCref(cref);
 	}
 
 	/***
 	 * Wakes up all tasks which are blocking. All tasks in this.transactionDAO
 	 * and this.blockOperations are waked up and added to the thread pool.
 	 */
 	private void wakeUpBlocking() {
 		for (OperationTask t : this.transactionDAO.take()) {
 			EventProcessingPool.getInstance().execute(t);
 			if (logger.isDebugEnabled()) {
 				logger.debug("wakeUpBlocking() waked up: " + t);
 			}
 		}
 		// wake up all blocking operations.
 		for (OperationTask t : this.blockingOperations.takeAll()) {
 			EventProcessingPool.getInstance().execute(t);
 			if (logger.isDebugEnabled()) {
 				logger.debug("wakeUpBlocking() waked up: " + t);
 			}
 		}
 	}
 
 	/***
 	 * 
 	 * {@inheritDoc}.
 	 */
 	public void rollback(Transaction txn) throws InvalidTransactionException,
 			TransactionLockException, AspectRescheduleException,
 			AspectNotOkException {
 		this.transactionDAO.writeLock().lock();
 		try {
 			c.rollback(txn);
 			this.wakeUpBlocking();
 		} finally {
 			this.transactionDAO.writeLock().unlock();
 		}
 	}
 
 	/***
 	 * 
 	 * {@inheritDoc}.
 	 */
 	public void commit(Transaction txn) throws TransactionLockException,
 			InvalidTransactionException, AspectRescheduleException,
 			AspectNotOkException {
 
 		this.transactionDAO.writeLock().lock();
 		this.transactionDAO.commit(txn);
 		try {
 			c.commit(txn);
 			this.wakeUpBlocking();
 		} finally {
 			this.transactionDAO.writeLock().unlock();
 		}
 	}
 
 	/***
 	 * 
 	 * {@inheritDoc}.
 	 */
 	public void updateTimeouts() {
 		if (logger.isDebugEnabled()) {
 			logger.debug("updateTimeouts() in Container: " + this.getCref());
 		}
 		this.rwLock.writeLock().lock();
 		try {
 			for (OperationTask task : this.blockingOperations.getAll()) {
 				task.updateTimeout();
 				if (task.timeoutExpired()) {
 					// remove the expired task.
 					this.blockingOperations.remove(task);
 					task.setResult(new TimeoutExpiredException());
 				}
 			}
 
 			for (OperationTask task : this.transactionDAO.getAll()) {
 				task.updateTimeout();
 				if (task.timeoutExpired()) {
 					// remove the expired task.
 					this.transactionDAO.remove(task);
 					task.setResult(new TimeoutExpiredException());
 				}
 			}
 		} finally {
 			if (this.rwLock != null) {
 				this.rwLock.writeLock().unlock();
 			}
 		}
 		if (logger.isDebugEnabled()) {
 			logger.debug("updateTimeouts() updated timeouts.");
 		}
 	}
 
 	/***
 	 * 
 	 * {@inheritDoc}.
 	 */
 	public String addAspects(List<IPoint> points, IAspect aspect,
 			Properties aspectContext) {
 		return this.c.addAspects(points, aspect, aspectContext);
 
 	}
 
 	/***
 	 * 
 	 * {@inheritDoc}.
 	 */
 	public void removeAspect(IPoint p, URI uri, Properties aspectContext) {
 		this.c.removeAspect(p, uri, aspectContext);
 	}
 
 	/***
 	 * 
 	 * {@inheritDoc}.
 	 */
 	public void addCoordinator(Class<? extends Selector> s, ICoordinator coord) {
 		this.c.addCoordinator(s, coord);
 
 	}
 
 	/***
 	 * {@inheritDoc}
 	 */
 	// @Override
 	public List<ICoordinator> getCoordinators() {
 		return this.c.getCoordinators();
 	}
 
 	/***
 	 * {@inheritDoc}
 	 */
 	public void destroy() {
 		for (OperationTask o : this.transactionDAO.take()) {
 			o.setResult(new InvalidContainerException(
 					"Container has been removed."));
 		}
 		// OPTMIZE if no copy is create a concurrent modification exception is
 		// thrown
 		for (OperationTask o : new ArrayList<OperationTask>(
 				this.blockingOperations.getAll())) {
 			o.setResult(new InvalidContainerException(
 					"Container has been removed."));
 		}
 	}
 
 	/***
 	 * {@inheritDoc}
 	 */
 	public int currentSize() {
 		return c.currentSize();
 	}
 }