package org.xvsm.internal;
   
   import java.util.ArrayList;
   import java.util.Collection;
   import java.util.HashMap;
   import java.util.List;
   import java.util.Map;
   import java.util.concurrent.ConcurrentHashMap;
   
  import org.apache.log4j.Logger;
  import org.xvsm.coordinators.RandomCoordinator;
  import org.xvsm.core.ContainerRef;
  import org.xvsm.core.Entry;
  import org.xvsm.interfaces.ICoordinator;
  import org.xvsm.interfaces.IExplicitCoordinator;
  import org.xvsm.interfaces.IImplicitCoordinator;
  import org.xvsm.interfaces.container.IContainer;
  import org.xvsm.interfaces.container.IContainerEngine;
  import org.xvsm.internal.exceptions.CannotShiftException;
  import org.xvsm.internal.exceptions.ContainerFullException;
  import org.xvsm.internal.exceptions.CountNotMetException;
  import org.xvsm.internal.exceptions.EntryLockedException;
  import org.xvsm.internal.exceptions.FatalException;
  import org.xvsm.internal.exceptions.NoSuchCoordinationTypeException;
  import org.xvsm.internal.exceptions.TransactionLockException;
  import org.xvsm.selectors.RandomSelector;
  import org.xvsm.selectors.Selector;
  import org.xvsm.transactions.Transaction;
  
  /***
   * @author Christian Schreiber, Michael Proestler
   * 
   */
  public class ContainerEngine implements IContainerEngine {
  
  	/***
  	 * The Logger.
  	 */
  	private static Logger logger = Logger.getLogger(ContainerEngine.class);
  
  	/***
  	 * Stores all implicit Coordination Types of this container. The key of the
  	 * map is the Class of the selector.
  	 */
  	private Map<Class<? extends Selector>, IImplicitCoordinator> implicitCoordinators = new ConcurrentHashMap<Class<? extends Selector>, IImplicitCoordinator>();
  
  	/***
  	 * Stores all explicit Coordination Types of this container. The key of the
  	 * map is the Class of the selector.
  	 */
  	private Map<Class<? extends Selector>, IExplicitCoordinator> explicitCoordinators = new ConcurrentHashMap<Class<? extends Selector>, IExplicitCoordinator>();
  
  	/***
  	 * The number of entries currently in the container.
  	 */
  	private int entryCount = 0;
  
  	/***
  	 * The size of this container.
  	 */
  	private int size;
  
  	/***
  	 * The {@link ContainerRef}.
  	 */
  	private ContainerRef cref;
  
  	/***
  	 * Creates a new unbounded Container. Random Coordination type is added to
  	 * the implicitCoordinationTypes.
  	 */
  	public ContainerEngine() {
  		this(IContainer.INFINITE_SIZE);
  	}
  
  	/***
  	 * Creates a new unbounded Container. Random Coordination type is added to
  	 * the implicitCoordinationTypes.
  	 * 
  	 * @param size
  	 *            the size of the container.
  	 */
  	public ContainerEngine(int size) {
  		this.setSize(size);
  		RandomCoordinator rc = new RandomCoordinator();
  		rc.setMaxContainerSize(size);
  		this.implicitCoordinators.put(RandomSelector.class, rc);
  	}
  
  	/***
  	 * {@inheritDoc}.
  	 */
  	public void commit(Transaction txn) throws TransactionLockException {
  		if (logger.isDebugEnabled()) {
  			logger.debug("commit(" + txn + ")");
  		}
  		// OPTIMIZE only commit modified coordinators!
  		// txn.getModifiedCoordinators is broken at the moment!!!
  		for (ICoordinator c : this.implicitCoordinators.values()) {
 			c.commit(txn);
 			txn.removeCoordinator(c);
 		}
 
 		for (ICoordinator c : this.explicitCoordinators.values()) {
 			c.commit(txn);
 			txn.removeCoordinator(c);
 		}
 		this.commitEntryLocks(txn);
 		if (logger.isDebugEnabled()) {
 			logger.debug("commit() transaction commited");
 		}
 	}
 
 	/***
 	 * {@inheritDoc}.
 	 */
 	public void commitSubTransaction(Transaction txn)
 			throws TransactionLockException {
 		if (logger.isDebugEnabled()) {
 			logger.debug("commit(" + txn + ")");
 		}
 		// OPTIMIZE only commit modified coordinators!
 		// txn.getModifiedCoordinators is broken at the moment!!!
 		for (ICoordinator c : this.implicitCoordinators.values()) {
 			c.commitSubTransaction(txn);
 			txn.removeCoordinator(c);
 		}
 
 		for (ICoordinator c : this.explicitCoordinators.values()) {
 			c.commitSubTransaction(txn);
 			txn.removeCoordinator(c);
 		}
 		this.commitEntryLocks(txn);
 		if (logger.isDebugEnabled()) {
 			logger.debug("commit() transaction commited");
 		}
 	}
 
 	/***
 	 * {@inheritDoc}.
 	 */
 	public void rollback(Transaction txn) throws TransactionLockException {
 		if (logger.isDebugEnabled()) {
 			logger.debug("rollback(" + txn + ")");
 		}
 		// OPTIMIZE only rollback modified coordinators!
 		// txn.getModifiedCoordinators is broken at the moment!!!
 		for (ICoordinator c : this.implicitCoordinators.values()) {
 			c.rollback(txn);
 			txn.removeCoordinator(c);
 		}
 
 		for (ICoordinator c : this.explicitCoordinators.values()) {
 			c.rollback(txn);
 			txn.removeCoordinator(c);
 		}
 		this.rollbackEntryLocks(txn);
 		if (logger.isDebugEnabled()) {
 			logger.debug("rollback() transaction rolled back");
 		}
 	}
 
 	/***
 	 * {@inheritDoc}.
 	 */
 	public int getSize() {
 		return this.size;
 	}
 
 	/***
 	 * {@inheritDoc}.
 	 */
 	public List<Entry> read(Transaction tx, List<Selector> selectors)
 			throws TransactionLockException, NoSuchCoordinationTypeException,
 			CountNotMetException {
 		if (logger.isDebugEnabled()) {
 			logger.debug("read(" + tx + ", " + selectors);
 		}
 
 		selectors = this.checkSelectors(selectors);
 
 		// apply the selectors
 		List<Entry> entries = null;
 		for (Selector s : selectors) {
 			entries = this.getCoordTypefromSelector(s.getClass()).read(tx, s,
 					entries);
 			for (Entry e : entries) {
 				try {
 					e.addReadLock(tx);
 					tx.addEntry(this.getCref(), e);
 				} catch (EntryLockedException e1) {
 					throw new TransactionLockException(e1.getReason());
 				}
 			}
 		}
 
 		if (logger.isDebugEnabled()) {
 			logger.debug("read(): result = " + entries);
 		}
 
 		return entries;
 	}
 
 	/***
 	 * {@inheritDoc}.
 	 */
 	public void write(Entry entry, Transaction tx)
 			throws ContainerFullException, TransactionLockException,
 			NoSuchCoordinationTypeException {
 		if (logger.isDebugEnabled()) {
 			logger.debug("write(" + entry + ", " + tx + ")");
 		}
 		entry.setSelectors(this.checkSelectors(entry.getSelectors()));
 
 		// is space left in the Container?
 		if (this.isFull()) {
 			throw new ContainerFullException(
 					"There is no free space in the container.");
 		}
 
 		List<ICoordinator> allImplCoordinators = new ArrayList<ICoordinator>(
 				this.implicitCoordinators.values());
 
 		// lock the entry
 		try {
 			entry.setWriteLock(tx);
 			tx.addEntry(this.getCref(), entry);
 		} catch (EntryLockedException e) {
 			throw new TransactionLockException(e.getReason());
 		}
 
 		// write the Entry with each selector
 		ICoordinator containerType;
 		for (Selector s : entry.getSelectors()) {
 			// Get the CoordinationType
 			containerType = this.getCoordTypefromSelector(s.getClass());
 			if (containerType == null) {
 				throw new NoSuchCoordinationTypeException(
 						" Can not find coordination type for selector "
 								+ s.getClass());
 			}
 			// write the entry and store the coordination type in the entry
 			if (!allImplCoordinators.contains(containerType)) {
 				containerType.write(entry, tx, s);
 				entry.addCoordinationTypes(containerType);
 				tx.addCoordinator(containerType);
 				// remember that this implicit coordinator has been executed.
 				allImplCoordinators.remove(containerType);
 			}
 
 			this.entryCount++;
 		}
 
 		for (ICoordinator c : allImplCoordinators) {
 			c.write(entry, tx, null);
 			entry.addCoordinationTypes(c);
 			tx.addCoordinator(c);
 		}
 
 		if (logger.isDebugEnabled()) {
 			logger.debug("write() entry written");
 		}
 	}
 
 	/***
 	 * {@inheritDoc}.
 	 */
 	public ICoordinator getCoordTypefromSelector(Class<? extends Selector> s)
 			throws NoSuchCoordinationTypeException {
 		ICoordinator coordType = this.implicitCoordinators.get(s);
 		if (coordType == null) {
 			coordType = this.explicitCoordinators.get(s);
 		}
 		if (coordType == null) {
 			throw new NoSuchCoordinationTypeException();
 
 		}
 		return coordType;
 	}
 
 	/***
 	 * Checks if there is space left in the container.
 	 * 
 	 * @return <code>true</code> if the container has at least one place left,
 	 *         otherwise <code>false</code>
 	 */
 	private boolean isFull() {
 		return this.size != IContainer.INFINITE_SIZE
 				&& this.size <= this.entryCount;
 	}
 
 	/***
 	 * Deletes one specific entry.
 	 * 
 	 * @param tx
 	 *            The transaction to use.
 	 * @param e
 	 *            The entry to delete.
 	 * @throws TransactionLockException
 	 *             if an entry or container could not be accessed because of
 	 *             another transactions lock.
 	 */
 	private void delete(Transaction tx, Entry e)
 			throws TransactionLockException {
 		if (logger.isDebugEnabled()) {
 			logger.debug("delete(" + tx + ", " + e + ")");
 		}
 
 		try {
 			e.setDeleteLock(tx);
 			tx.addEntry(this.getCref(), e);
 		} catch (EntryLockedException e1) {
 			throw new TransactionLockException(e1.getReason());
 		}
 
 		for (ICoordinator ct : e.getCoordinationType()) {
 			// delete the entry from all coordination types which have a
 			// reference to it. This means no reference is left and the
 			// entry is deleted.
 			ct.delete(tx, e);
 
 		}
 		if (logger.isDebugEnabled()) {
 			logger.debug("delete() entry deleted");
 		}
 
 		this.entryCount--;
 	}
 
 	/***
 	 * Checks if the selectors are not null. If selectors is null or empty an
 	 * array with one Random Selector will be returned.
 	 * 
 	 * @param selectors
 	 *            the selectors to check.
 	 * @return the checked selectors.
 	 */
 	private List<Selector> checkSelectors(List<Selector> selectors) {
 		// if no selector is given use the default => RandomSelector(1).
 		if (selectors == null || selectors.size() == 0
 				|| selectors.get(0) == null) {
 			selectors = new ArrayList<Selector>();
 			selectors.add(new RandomSelector(1));
 		}
 		return selectors;
 	}
 
 	/***
 	 * {@inheritDoc}.
 	 */
 	public List<Entry> shift(Entry entry, Transaction tx)
 			throws TransactionLockException, NoSuchCoordinationTypeException {
 		if (logger.isDebugEnabled()) {
 			logger.debug("shift(" + entry + ", " + tx);
 		}
 
 		// lock the entry
 		try {
 			entry.setWriteLock(tx);
 			tx.addEntry(this.getCref(), entry);
 		} catch (EntryLockedException e) {
 			throw new TransactionLockException(e.getReason());
 		}
 
 		List<Entry> shifted = new ArrayList<Entry>();
 
 		entry.setSelectors(this.checkSelectors(entry.getSelectors()));
 
 		// write the Entry with each selector
 		ICoordinator coordinator;
 
 		// TODO Optimize this.
 		Map<Selector, ICoordinator> failed = new HashMap<Selector, ICoordinator>();
 		for (Selector s : entry.getSelectors()) {
 
 			// Get the CoordinationType
 			// execute the explicit coordinators. If one fails remember it and
 			// execute it after the implicit coordinators.
 			coordinator = this.getCoordTypefromSelector(s.getClass());
 			try {
 				if (coordinator == null) {
 					throw new NoSuchCoordinationTypeException(
 							" Can not find coordination type for selector "
 									+ s.getClass());
 				}
 				if (coordinator instanceof IExplicitCoordinator) {
 					Entry e = coordinator.shift(entry, tx, s);
 					// if an entry has been shifted delete it.
 					if (e != null) {
 						this.delete(tx, e);
 						shifted.add(e);
 					}
 					// store that this entry is in the coordinator.
 					entry.addCoordinationTypes(coordinator);
 				}
 			} catch (Exception e) {
 				// remember
 				failed.put(s, coordinator);
 			}
 		}
 
 		try {
 			// execute all implicit coordinators.
 			Collection<IImplicitCoordinator> implCoordinators = new ArrayList<IImplicitCoordinator>(
 					this.implicitCoordinators.values());
 			for (Selector s : entry.getSelectors()) {
 				coordinator = this.getCoordTypefromSelector(s.getClass());
 				if (coordinator instanceof IImplicitCoordinator) {
 					if (coordinator == null) {
 						throw new NoSuchCoordinationTypeException(
 								" Can not find coordination type for selector "
 										+ s.getClass());
 					}
 					Entry e = coordinator.shift(entry, tx, s);
 					// if an entry has been shifted delete it.
 					if (e != null) {
 						this.delete(tx, e);
 						shifted.add(e);
 					}
 					// store that this entry is in the coordinator.
 					entry.addCoordinationTypes(coordinator);
 					implCoordinators.remove(coordinator);
 				}
 			}
 
 			// run the remaining (those without a selector) implicit
 			// coordinators
 			for (IImplicitCoordinator c : new ArrayList<IImplicitCoordinator>(
 					implCoordinators)) {
 				Entry e = c.shift(entry, tx, null);
 				// if an entry has been shifted delete it.
 				if (e != null) {
 					this.delete(tx, e);
 					shifted.add(e);
 				}
 				// store that this entry is in the coordinator.
 				entry.addCoordinationTypes(c);
 				implCoordinators.remove(c);
 			}
 
 			// execute the failed explicit coordinators
 			for (Selector s : failed.keySet()) {
 				ICoordinator c = failed.get(s);
 				Entry e = c.shift(entry, tx, s);
 				// if an entry has been shifted delete it.
 				if (e != null) {
 					this.delete(tx, e);
 					shifted.add(e);
 				}
 				// store that this entry is in the coordinator.
 				entry.addCoordinationTypes(c);
 				implCoordinators.remove(c);
 			}
 		} catch (CannotShiftException e) {
 			// this exception should not happen here.
 			throw new FatalException(e);
 		}
 
 		this.entryCount++;
 
 		if (logger.isDebugEnabled()) {
 			logger.debug("shift() shifted entry: " + shifted);
 		}
 
 		return shifted;
 	}
 
 	/***
 	 * {@inheritDoc}.
 	 */
 	public List<Entry> take(Transaction tx, List<Selector> selectors)
 			throws TransactionLockException, NoSuchCoordinationTypeException,
 			CountNotMetException {
 
 		if (logger.isDebugEnabled()) {
 			logger.debug("take(" + tx + ", " + selectors + ")");
 		}
 
 		// OPTIMIZE optimize this. without the copy a
 		// ConcurrentModificationException is thrown.
 		List<Entry> entries = new ArrayList<Entry>(this.read(tx, selectors));
 
 		for (Entry e : entries) {
 			this.delete(tx, e);
 		}
 
 		if (logger.isDebugEnabled()) {
 			logger.debug("take() entries taken: " + entries);
 		}
 		return entries;
 	}
 
 	/***
 	 * Sets the size of this container.
 	 * 
 	 * @param size
 	 *            the new container size.
 	 */
 	private void setSize(int size) {
 		this.size = size;
 	}
 
 	/***
 	 * Get the {@link ContainerRef} of the container.
 	 * 
 	 * @return the {@link ContainerRef}
 	 */
 	// @Override
 	public ContainerRef getCref() {
 		return this.cref;
 	}
 
 	/***
 	 * Set the {@link ContainerRef} of the container.
 	 * 
 	 * @param cref
 	 *            the new {@link ContainerRef}
 	 */
 	// @Override
 	public void setCref(ContainerRef cref) {
 		this.cref = cref;
 
 		for (IImplicitCoordinator c : this.implicitCoordinators.values()) {
 			c.setCref(cref);
 		}
 		for (IExplicitCoordinator c : this.explicitCoordinators.values()) {
 			c.setCref(cref);
 		}
 	}
 
 	/***
 	 * 
 	 * {@inheritDoc}.
 	 */
 	// @Override
 	public void addCoordinator(Class<? extends Selector> s, ICoordinator c) {
 		c.setMaxContainerSize(this.size);
 		c.setCref(this.getCref());
 		if (c instanceof IImplicitCoordinator) {
 			this.implicitCoordinators.put(s, (IImplicitCoordinator) c);
 			return;
 		}
 		if (c instanceof IExplicitCoordinator) {
 			this.explicitCoordinators.put(s, (IExplicitCoordinator) c);
 			return;
 		}
 		throw new IllegalArgumentException(
 				"ICoordinator not allowed as superclass.");
 	}
 
 	/***
 	 * {@inheritDoc}
 	 * 
 	 */
 	// @Override
 	public List<ICoordinator> getCoordinators() {
 		List<ICoordinator> list = new ArrayList<ICoordinator>();
 		list.addAll(this.implicitCoordinators.values());
 		list.addAll(this.explicitCoordinators.values());
 		return list;
 	}
 
 	/***
 	 * Updates the lock on all entries which have been modified in this
 	 * container with tx.<br>
 	 * The locks will be set to the father transaction of tx or to
 	 * <code>null</code> if tx has no father.
 	 * 
 	 * @param tx
 	 *            the transaction to commit.
 	 * @throws TransactionLockException
 	 *             thrown if the lock can not be acquired.
 	 */
 	protected void commitEntryLocks(Transaction tx)
 			throws TransactionLockException {
 		if (logger.isDebugEnabled()) {
 			logger.debug("commitEntryLocks(" + tx + ")");
 		}
 		try {
 			for (Entry e : tx.getModifiedEntries(this.getCref())) {
 				if (e.hasDeleteLock(tx)) {
 					e.setDeleteLock(tx.getFather());
 				}
 				if (e.hasWriteLock(tx)) {
 					e.setWriteLock(tx.getFather());
 				}
 				e.removeReadLock(tx);
 				e.addReadLock(tx.getFather());
 				if (tx.getFather() != null) {
 					tx.getFather().addEntry(this.getCref(), e);
 				}
 			}
 		} catch (EntryLockedException e) {
 			throw new TransactionLockException(e.getReason());
 		}
 		if (logger.isDebugEnabled()) {
 			logger.debug("commitEntryLocks() commited entries for coordinator "
 					+ this.getClass());
 		}
 	}
 
 	/***
 	 * Updates the lock on all entries which have been modified in this
 	 * container with tx.<br>
 	 * The locks will be <code>null</code>.
 	 * 
 	 * @param tx
 	 *            the transaction to rollback.
 	 * @throws TransactionLockException
 	 *             thrown if the lock can not be aquired.
 	 */
 	protected void rollbackEntryLocks(Transaction tx)
 			throws TransactionLockException {
 		if (logger.isDebugEnabled()) {
 			logger.debug("rollbackEntryLocks(" + tx + ")");
 		}
 		try {
 
 			for (Entry e : tx.getModifiedEntries(this.getCref())) {
 				if (e.hasDeleteLock(tx)) {
 					e.setDeleteLock(null);
 				}
 				if (e.hasWriteLock(tx)) {
 					e.setWriteLock(null);
 				}
 				e.removeReadLock(tx);
 				tx.removeEntry(this.getCref(), e);
 			}
 
 		} catch (EntryLockedException e) {
 			throw new TransactionLockException(e.getReason());
 		}
 		if (logger.isDebugEnabled()) {
 			logger.debug("rollbackEntryLocks() locks released for coordinator "
 					+ this.getClass());
 		}
 	}
 
 	/***
 	 * 
 	 * {@inheritDoc}.
 	 */
 	public int currentSize() {
 		return this.entryCount;
 	}
 
 }