package org.xvsm.coordinators;
   
   import java.util.ArrayList;
   import java.util.HashMap;
   import java.util.List;
   import java.util.Map;
   import java.util.Properties;
   
   import org.apache.log4j.Logger;
  import org.xvsm.core.AtomicEntry;
  import org.xvsm.core.Entry;
  import org.xvsm.core.Tuple;
  import org.xvsm.interfaces.IExplicitCoordinator;
  import org.xvsm.interfaces.container.IContainer;
  import org.xvsm.internal.exceptions.CannotShiftException;
  import org.xvsm.internal.exceptions.ContainerFullException;
  import org.xvsm.internal.exceptions.CountNotMetException;
  import org.xvsm.internal.exceptions.FatalException;
  import org.xvsm.internal.exceptions.TransactionLockException;
  import org.xvsm.selectors.KeySelector;
  import org.xvsm.selectors.Selector;
  import org.xvsm.transactions.Transaction;
  
  /***
   * @author Christian Schreiber, Michael Proestler
   * 
   */
  public class KeyCoordinator extends IExplicitCoordinator {
  
  	/***
  	 * The logger.
  	 */
  	private static Logger logger = Logger.getLogger(KeyCoordinator.class);
  
  	/***
  	 * The SerialVersionUID of this Class.
  	 */
  	private static final long serialVersionUID = 3027037571449489139L;
  
  	/***
  	 * Contains all entries which have been written with a KeySelector.
  	 */
  	private Map<String, Map<Object, Entry>> entries;
  
  	/***
  	 * The maximum size of the container where this KeyCoordinator belongs to.
  	 */
  	private int maxSize;
  
  	/***
  	 * The amount of entries currently managed by this coordinator.
  	 */
  	private int currentSize;
  
  	/***
  	 * Contains entries and selectors which have been deleted. This information
  	 * is necessary to rollback a transaction.
  	 */
  	private Map<Transaction, Map<Entry, KeySelector>> deletedEntries = new HashMap<Transaction, Map<Entry, KeySelector>>();
  
  	private Map<String, Class> typeMap = new HashMap<String, Class>();
  
  	/***
  	 * We need to store which entry has been written with which selector.
  	 * Otherwise we need to iterate over all entries on delete.
  	 */
  	private Map<Entry, KeySelector> reverseMap = new HashMap<Entry, KeySelector>();
  
  	public static class KeyType {
  		public String keyName;
  		public Class keyType;
  
  		public KeyType(String keyName, Class keyType) {
  			this.keyName = keyName;
  			this.keyType = keyType;
  		}
  	}
  
  	/***
  	 * Default Constructor.
  	 * 
  	 */
  	public KeyCoordinator() {
  		this.entries = new HashMap<String, Map<Object, Entry>>();
  	}
  
  	public KeyCoordinator(KeyType... keytypes) {
  		this();
  		for (KeyType kt : keytypes) {
  			this.entries.put(kt.keyName, new HashMap<Object, Entry>());
  			this.typeMap.put(kt.keyName, kt.keyType);
  		}
  	}
  
  	/***
  	 * {@inheritDoc}.
  	 */
  	public void commit(Transaction tx) throws TransactionLockException {
  		// remove the information about the deleted entries.
 		this.deletedEntries.remove(tx);
 	}
 
 	/***
 	 * {@inheritDoc}
 	 * 
 	 * @throws TransactionLockException
 	 */
 	@Override
 	public void commitSubTransaction(Transaction tx)
 			throws TransactionLockException {
 		Map<Entry, KeySelector> deletedUnderChildTx = deletedEntries.get(tx);
 		if (deletedUnderChildTx != null) {
 			deletedEntries.put(tx.getFather(), deletedUnderChildTx);
 		}
 	}
 
 	/***
 	 * {@inheritDoc}.
 	 */
 	public void delete(Transaction tx, Entry e) throws TransactionLockException {
 		KeySelector ks = this.reverseMap.get(e);
 		if (logger.isDebugEnabled()) {
 			logger.debug("KeySelector: " + ks);
 		}
 		if (ks != null) {
 			this.remove(e, ks, tx);
 			// remove the entry from the reverse map because it is no longer
 			// needed.
 			this.reverseMap.remove(e);
 		}
 	}
 
 	/***
 	 * {@inheritDoc}.
 	 */
 	public List<Entry> read(Transaction tx, Selector selector,
 			List<Entry> centries) throws TransactionLockException,
 			CountNotMetException {
 		KeySelector<?> ks = (KeySelector<?>) selector;
 
 		// there can only be one entry read (there can not be more then one
 		// entry with the same key). therefore we ignore the count of
 		// the selector.
 		Map<Object, Entry> keyMap = this.entries.get(ks.getKeyName());
 		Entry e;
 		if (keyMap == null) {
 			e = null;
 		} else {
 			e = keyMap.get(ks.getKeyValue());
 		}
 		if (e == null || !e.canBeRead(tx)) {
 			throw new CountNotMetException(
 					"No entry with this key found. Key: " + ks.getKeyName()
 							+ " : " + ks.getKeyValue());
 		}
 
 		List<Entry> retVal = new ArrayList<Entry>();
 		// is e in the preselected entries?
 		if (centries != null) {
 			for (Entry en : centries) {
 				if (e.equals(en)) {
 					// ok, the entry is in the preselected entries.
 					// super.readLockEntry(e, tx);
 					retVal.add(en);
 					break;
 				}
 			}
 			if (retVal.size() == 0) {
 				throw new CountNotMetException(
 						"The entry was not in the list of the preselected entries.");
 			}
 		} else {
 			// super.readLockEntry(e, tx);
 			retVal.add(e);
 		}
 
 		return retVal;
 	}
 
 	/***
 	 * {@inheritDoc}.
 	 */
 	public void rollback(Transaction tx) throws TransactionLockException {
 		// remove entries which have been written under tx:
 		for (Entry e : tx.getModifiedEntries(this.getCref())) {
 			KeySelector s = this.reverseMap.get(e);
 			if (s != null) {
 				Map<Object, Entry> map = this.entries.get(s.getKeyName());
 				Entry entry = map.get(s.getKeyValue());
 				if (entry.removeOnRollback(tx)) {
 					map.remove(s.getKeyValue());
 				}
 			}
 		}
 
 		// read entries which have been deleted under tx:
 		Map<Entry, KeySelector> deleted = this.deletedEntries.get(tx);
 		if (deleted != null) {
 			for (Map.Entry<Entry, KeySelector> x : deleted.entrySet()) {
 				Map<Object, Entry> keymap = this.entries.get(x.getValue()
 						.getKeyName());
 				if (keymap == null) {
 					keymap = new HashMap<Object, Entry>();
 					this.entries.put(x.getValue().getKeyName(), keymap);
 				}
 				keymap.put(x.getValue().getKeyValue(), x.getKey());
 				this.currentSize++;
 			}
 		}
 	}
 
 	/***
 	 * {@inheritDoc}.
 	 */
 	public void setMaxContainerSize(int size) {
 		this.maxSize = size;
 	}
 
 	/***
 	 * {@inheritDoc}.
 	 */
 	public Entry shift(Entry e, Transaction tx, Selector s)
 			throws TransactionLockException, CannotShiftException {
 		KeySelector ks = (KeySelector) s;
 		// get the map for this key. If it does not exist create it.
 		Map<Object, Entry> keyMap = this.entries.get(ks.getKeyName());
 		if (keyMap == null) {
 			throw new FatalException("key \"" + ks.getKeyName()
 					+ "\" does not exist");
 		}
 
 		// is there already an entry at this position?
 		Entry shifted = keyMap.get(ks.getKeyValue());
 		if (shifted != null && !shifted.hasDeleteLock(tx)) {
 			if (this.currentSize > this.maxSize
 					&& this.maxSize != IContainer.INFINITE_SIZE) {
 				throw new CannotShiftException(
 						"Can not decide which Entry has to be deleted");
 			} else {
 				// delete the entry.
 				this.remove(shifted, ks, tx);
 			}
 		}
 		// store the new entry.
 		keyMap.put(ks.getKeyValue(), e);
 		this.reverseMap.put(e, ks);
 		this.currentSize++;
 		return shifted;
 	}
 
 	/***
 	 * Removes the entry and stores it in this.deletedEntries.
 	 * 
 	 * @param e
 	 *            the entry to remove
 	 * @param s
 	 *            the selector for the entry.
 	 * @param tx
 	 *            the Transaction.
 	 */
 	private void remove(Entry e, KeySelector s, Transaction tx) {
 		Map<Entry, KeySelector> deleted = this.deletedEntries.get(tx);
 		if (deleted == null) {
 			deleted = new HashMap<Entry, KeySelector>();
 			this.deletedEntries.put(tx, deleted);
 		}
 		deleted.put(e, s);
 		if (e.getId().equals(
 				this.entries.get(s.getKeyName()).get(s.getKeyValue()).getId())) {
 			// if the entry at the position is not equals to the entry that has
 			// been shifted and at the position is now a new entry which must
 			// not be deleted.
 			this.entries.get(s.getKeyName()).remove(s.getKeyValue());
 		}
 
 		this.currentSize--;
 	}
 
 	/***
 	 * {@inheritDoc}.
 	 */
 	public void write(Entry e, Transaction tx, Selector s)
 			throws ContainerFullException, TransactionLockException {
 		KeySelector<?> ks = (KeySelector<?>) s;
 
 		// is there place left?
 		if (this.currentSize == this.maxSize
 				&& this.maxSize != IContainer.INFINITE_SIZE) {
 			throw new ContainerFullException();
 		}
 
 		// get the map for this key. If it does not exist create it.
 		Map<Object, Entry> keyMap = this.entries.get(ks.getKeyName());
 		if (keyMap == null) {
 			throw new FatalException("key \"" + ks.getKeyName()
 					+ "\" does not exist");
 		}
 
 		// is there already an old entry on this position?
 		Entry olde = keyMap.get(ks.getKeyValue());
 		if (olde != null) {
 			if (!olde.hasDeleteLock(tx)) {
 				// block.
 				throw new ContainerFullException();
 			} else {
 				// there is an entry at this positition which is marked as
 				// deleted.
 				this.remove(olde, ks, tx);
 			}
 		}
 		keyMap.put(ks.getKeyValue(), e);
 		this.reverseMap.put(e, ks);
 		this.currentSize++;
 	}
 
 	/***
 	 * {@inheritDoc}.
 	 */
 	@Override
 	public Class<? extends Selector> getDefaultSelector() {
 		return KeySelector.class;
 	}
 
 	@Override
 	public Properties getProperties() {
 		Tuple t = new Tuple();
 		for (String keyname : this.entries.keySet()) {
 			t.add(new Tuple(new AtomicEntry<String>(keyname),
 					new AtomicEntry<String>(typeMap.get(keyname)
 							.getSimpleName().toLowerCase())));
 		}
 		this.properties.put("keys", t);
 		return this.properties;
 	}
 
 	@Override
 	public void setProperties(Properties properties) {
 		try {
 			Tuple keys = (Tuple) properties.get("keys");
 			for (Entry e : keys) {
 				Tuple t = (Tuple) e;
 				String keyName = (String) ((AtomicEntry<?>) t.getEntryAt(0))
 						.getValue();
 
 				String keyType = (String) ((AtomicEntry<?>) t.getEntryAt(1))
 						.getValue();
 
 				Class<?> c = Class.forName("java.lang."
 						+ keyType.substring(0, 1).toUpperCase()
 						+ keyType.substring(1, keyType.length()));
 
 				this.typeMap.put(keyName, c);
 
 				this.entries.put(keyName, new HashMap<Object, Entry>());
 			}
 		} catch (ClassNotFoundException e) {
 			throw new FatalException(e);
 		}
 	}
 }