/*
    * Licensed to the Apache Software Foundation (ASF) under one or more
    * contributor license agreements.  See the NOTICE file distributed with
    * this work for additional information regarding copyright ownership.
    * The ASF licenses this file to You under the Apache License, Version 2.0
    * (the "License"); you may not use this file except in compliance with
    * the License.  You may obtain a copy of the License at
    *
    *     http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
  package org.apache.commons.transaction.memory;
  
  import java.io.PrintWriter;
  import java.util.HashSet;
  import java.util.Iterator;
  import java.util.Map;
  import java.util.Set;
  import java.util.Collections;
  
  import org.apache.commons.transaction.locking.ReadWriteLock;
  import org.apache.commons.transaction.util.LoggerFacade;
  import org.apache.commons.transaction.util.PrintWriterLogger;
  
  /**
   * Wrapper that adds transactional control to all kinds of maps that implement the {@link Map} interface. By using
   * a naive optimistic transaction control this wrapper has better isolation than {@link TransactionalMapWrapper}, but
   * may also fail to commit. 
   *  
   * <br>
   * Start a transaction by calling {@link #startTransaction()}. Then perform the normal actions on the map and
   * finally either call {@link #commitTransaction()} to make your changes permanent or {@link #rollbackTransaction()} to
   * undo them.
   * <br>
   * <em>Caution:</em> Do not modify values retrieved by {@link #get(Object)} as this will circumvent the transactional mechanism.
   * Rather clone the value or copy it in a way you see fit and store it back using {@link #put(Object, Object)}.
   * <br>
   * <em>Note:</em> This wrapper guarantees isolation level <code>SERIALIZABLE</code>.
   * <br>
   * <em>Caution:</em> This implementation might be slow when large amounts of data is changed in a transaction as much references will need to be copied around.
   * 
   * @version $Id: OptimisticMapWrapper.java 493628 2007-01-07 01:42:48Z joerg $
   * @see TransactionalMapWrapper
   * @see PessimisticMapWrapper
   */
  public class OptimisticMapWrapper extends TransactionalMapWrapper {
  
      protected static final int COMMIT_TIMEOUT = 1000 * 60; // 1 minute
      protected static final int ACCESS_TIMEOUT = 1000 * 30; // 30 seconds
  
      protected Set activeTransactions;
  
      protected LoggerFacade logger;
  
      protected ReadWriteLock commitLock;
  
      /**
       * Creates a new optimistic transactional map wrapper. Temporary maps and sets to store transactional
       * data will be instances of {@link java.util.HashMap} and {@link java.util.HashSet}. 
       * 
       * @param wrapped map to be wrapped
       */
      public OptimisticMapWrapper(Map wrapped) {
          this(wrapped, new HashMapFactory(), new HashSetFactory());
      }
  
      /**
       * Creates a new optimistic transactional map wrapper. Temporary maps and sets to store transactional
       * data will be created and disposed using {@link MapFactory} and {@link SetFactory}.
       * 
       * @param wrapped map to be wrapped
       * @param mapFactory factory for temporary maps
       * @param setFactory factory for temporary sets
       */
      public OptimisticMapWrapper(Map wrapped, MapFactory mapFactory, SetFactory setFactory) {
          this(wrapped, mapFactory, setFactory, new PrintWriterLogger(new PrintWriter(System.out),
                  OptimisticMapWrapper.class.getName(), false));
      }
  
      /**
       * Creates a new optimistic transactional map wrapper. Temporary maps and sets to store transactional
       * data will be created and disposed using {@link MapFactory} and {@link SetFactory}.
       * 
       * @param wrapped map to be wrapped
       * @param mapFactory factory for temporary maps
       * @param setFactory factory for temporary sets
       * @param logger
       *            generic logger used for all kinds of logging
       */
      public OptimisticMapWrapper(Map wrapped, MapFactory mapFactory, SetFactory setFactory, LoggerFacade logger) {
          super(wrapped, mapFactory, setFactory);
          activeTransactions = Collections.synchronizedSet(new HashSet());
          this.logger = logger;
          commitLock = new ReadWriteLock("COMMIT", logger);
     }
 
     public void startTransaction() {
         if (getActiveTx() != null) {
             throw new IllegalStateException(
                 "Active thread " + Thread.currentThread() + " already associated with a transaction!");
         }
         CopyingTxContext context = new CopyingTxContext();
         activeTransactions.add(context);
         setActiveTx(context);
     }
 
     public void rollbackTransaction() {
         TxContext txContext = getActiveTx();
         super.rollbackTransaction();
         activeTransactions.remove(txContext);
     }
 
     public void commitTransaction() throws ConflictException {
         commitTransaction(false);
     }
 
     public void commitTransaction(boolean force) throws ConflictException {
         TxContext txContext = getActiveTx();
         
         if (txContext == null) {
             throw new IllegalStateException(
                 "Active thread " + Thread.currentThread() + " not associated with a transaction!");
         }
 
         if (txContext.status == STATUS_MARKED_ROLLBACK) {
             throw new IllegalStateException("Active thread " + Thread.currentThread() + " is marked for rollback!");
         }
         
         try {
             // in this final commit phase we need to be the only one access the map
             // to make sure no one adds an entry after we checked for conflicts
             commitLock.acquireWrite(txContext, COMMIT_TIMEOUT);
 
             if (!force) {
                 Object conflictKey = checkForConflicts();
                 if (conflictKey != null) {
                     throw new ConflictException(conflictKey);
                 }
             }
     
             activeTransactions.remove(txContext);
             copyChangesToConcurrentTransactions();
             super.commitTransaction();
             
         } catch (InterruptedException e) {
             // XXX a bit dirty ;)
             throw new ConflictException(e);
         } finally {
             commitLock.release(txContext);
         }
     }
 
     // TODO: Shouldn't we return a collection rather than a single key here?
     public Object checkForConflicts() {
         CopyingTxContext txContext = (CopyingTxContext) getActiveTx();
 
         Set keys = txContext.changedKeys();
         Set externalKeys = txContext.externalChangedKeys();
 
         for (Iterator it2 = keys.iterator(); it2.hasNext();) {
             Object key = it2.next();
             if (externalKeys.contains(key)) {
                 return key;
             }
         }
         return null;
     }
 
     protected void copyChangesToConcurrentTransactions() {
         CopyingTxContext thisTxContext = (CopyingTxContext) getActiveTx();
 
         synchronized (activeTransactions) {
             for (Iterator it = activeTransactions.iterator(); it.hasNext();) {
                 CopyingTxContext otherTxContext = (CopyingTxContext) it.next();
 
                 // no need to copy data if the other transaction does not access global map anyway
                 if (otherTxContext.cleared)
                     continue;
 
                 if (thisTxContext.cleared) {
                     // we will clear everything, so we have to copy everything before
                     otherTxContext.externalChanges.putAll(wrapped);
                 } else // no need to check if we have already copied everthing
                 {
                     for (Iterator it2 = thisTxContext.changes.entrySet().iterator(); it2.hasNext();) {
                         Map.Entry entry = (Map.Entry) it2.next();
                         Object value = wrapped.get(entry.getKey());
                         if (value != null) {
                             // undo change
                             otherTxContext.externalChanges.put(entry.getKey(), value);
                         } else {
                             // undo add
                             otherTxContext.externalDeletes.add(entry.getKey());
                         }
                     }
 
                     for (Iterator it2 = thisTxContext.deletes.iterator(); it2.hasNext();) {
                         // undo delete
                         Object key = it2.next();
                         Object value = wrapped.get(key);
                         otherTxContext.externalChanges.put(key, value);
                     }
                 }
             }
         }
     }
 
     public class CopyingTxContext extends TxContext {
         protected Map externalChanges;
         protected Map externalAdds;
         protected Set externalDeletes;
 
         protected CopyingTxContext() {
             super();
             externalChanges = mapFactory.createMap();
             externalDeletes = setFactory.createSet();
             externalAdds = mapFactory.createMap();
         }
 
         protected Set externalChangedKeys() {
             Set keySet = new HashSet();
             keySet.addAll(externalDeletes);
             keySet.addAll(externalChanges.keySet());
             keySet.addAll(externalAdds.keySet());
             return keySet;
         }
 
         protected Set changedKeys() {
             Set keySet = new HashSet();
             keySet.addAll(deletes);
             keySet.addAll(changes.keySet());
             keySet.addAll(adds.keySet());
             return keySet;
         }
 
         protected Set keys() {
             try {
                 commitLock.acquireRead(this, ACCESS_TIMEOUT);
                 Set keySet = super.keys();
                 keySet.removeAll(externalDeletes);
                 keySet.addAll(externalAdds.keySet());
                 return keySet;
             } catch (InterruptedException e) {
                 return null;
             } finally {
                 commitLock.release(this);
             }
         }
 
         protected Object get(Object key) {
             try {
                 commitLock.acquireRead(this, ACCESS_TIMEOUT);
 
                 if (deletes.contains(key)) {
                     // reflects that entry has been deleted in this tx 
                     return null;
                 }
     
                 Object changed = changes.get(key);
                 if (changed != null) {
                     return changed;
                 }
     
                 Object added = adds.get(key);
                 if (added != null) {
                     return added;
                 }
     
                 if (cleared) {
                     return null;
                 } else {
                     if (externalDeletes.contains(key)) {
                         // reflects that entry has been deleted in this tx 
                         return null;
                     }
     
                     changed = externalChanges.get(key);
                     if (changed != null) {
                         return changed;
                     }
     
                     added = externalAdds.get(key);
                     if (added != null) {
                         return added;
                     }
     
                     // not modified in this tx
                     return wrapped.get(key);
                 }
             } catch (InterruptedException e) {
                 return null;
             } finally {
                 commitLock.release(this);
             }
         }
 
         protected void put(Object key, Object value) {
             try {
                 commitLock.acquireRead(this, ACCESS_TIMEOUT);
                 super.put(key, value);
             } catch (InterruptedException e) {
             } finally {
                 commitLock.release(this);
             }
         }
 
         protected void remove(Object key) {
             try {
                 commitLock.acquireRead(this, ACCESS_TIMEOUT);
                 super.remove(key);
             } catch (InterruptedException e) {
             } finally {
                 commitLock.release(this);
             }
         }
 
         protected int size() {
             try {
                 commitLock.acquireRead(this, ACCESS_TIMEOUT);
                 int size = super.size();
     
                 size -= externalDeletes.size();
                 size += externalAdds.size();
     
                 return size;
             } catch (InterruptedException e) {
                 return -1;
             } finally {
                 commitLock.release(this);
             }
         }
 
         protected void clear() {
             try {
                 commitLock.acquireRead(this, ACCESS_TIMEOUT);
                 super.clear();
                 externalDeletes.clear();
                 externalChanges.clear();
                 externalAdds.clear();
             } catch (InterruptedException e) {
             } finally {
                 commitLock.release(this);
             }
         }
 
         protected void merge() {
             try {
                 commitLock.acquireRead(this, ACCESS_TIMEOUT);
                 super.merge();
             } catch (InterruptedException e) {
             } finally {
                 commitLock.release(this);
             }
         }
 
         protected void dispose() {
             try {
                 commitLock.acquireRead(this, ACCESS_TIMEOUT);
                 super.dispose();
                 setFactory.disposeSet(externalDeletes);
                 externalDeletes = null;
                 mapFactory.disposeMap(externalChanges);
                 externalChanges = null;
                 mapFactory.disposeMap(externalAdds);
                 externalAdds = null;
             } catch (InterruptedException e) {
             } finally {
                 commitLock.release(this);
             }
         }
 
         protected void finalize() throws Throwable {
             activeTransactions.remove(this);
             super.finalize();
         }
     }
 }