import junit.framework.TestCase;

import org.infinispan.AdvancedCache;
import org.infinispan.Cache;
import org.infinispan.configuration.cache.ConfigurationBuilder;
import org.infinispan.configuration.global.GlobalConfigurationBuilder;
import org.infinispan.configuration.parsing.ConfigurationBuilderHolder;
import org.infinispan.manager.DefaultCacheManager;
import org.infinispan.manager.EmbeddedCacheManager;
import org.infinispan.transaction.TransactionMode;
import org.infinispan.transaction.lookup.TransactionManagerLookup;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicReference;

import javax.transaction.TransactionManager;

public class TestLocking extends TestCase
{

   /**
    * @param args
    */
   public void testLocking() throws Exception
   {
      TransactionManagerLookup tml = new TransactionManagerLookup()
      {
         public TransactionManager getTransactionManager() throws Exception
         {
            return com.arjuna.ats.jta.TransactionManager.transactionManager();
         }
      };

      ConfigurationBuilderHolder holder = new ConfigurationBuilderHolder();
      GlobalConfigurationBuilder configBuilder = holder.getGlobalConfigurationBuilder();
      ConfigurationBuilder cb = holder.getDefaultConfigurationBuilder();
      cb.transaction().transactionManagerLookup(tml).transactionMode(TransactionMode.TRANSACTIONAL);
      cb.locking().lockAcquisitionTimeout(5000).useLockStriping(true).concurrencyLevel(512);

      EmbeddedCacheManager manager = new DefaultCacheManager(configBuilder.build(), cb.build());

      int writersPerCategory = 1;//5;
      final CountDownLatch startSignal = new CountDownLatch(1);
      final CountDownLatch endSignal = new CountDownLatch(2 * writersPerCategory);
      Cache<CacheKey, String> c = manager.getCache();
      final AdvancedCache<CacheKey, String> cache = (AdvancedCache<CacheKey, String>)c;

      final CacheKey key1 =
         new CacheKey("5486a755c0a80104686c4d3665a6c45c[http://www.jcp.org/jcr/1.0]primaryType:1PROPERTY");
      System.out.println("key = " + key1 + " lock id = " + cache.getLockManager().getLockId(key1));
      final CacheKey key2 = new CacheKey("548485f9c0a80104017d3a6ced75231a");
      System.out.println("key = " + key2 + " lock id = " + cache.getLockManager().getLockId(key2));
      final CacheKey key3 = new CacheKey("548485f9c0a80104017d3a6ced75231a[]rootNode1:1NODE");
      System.out.println("key = " + key3 + " lock id = " + cache.getLockManager().getLockId(key3));

      final AtomicReference<Exception> exception = new AtomicReference<Exception>();
      
      for (int i = 0; i < writersPerCategory; i++)
      {
         Thread t = new Thread("Writer Type 0 Index " + i)
         {
            public void run()
            {
               try
               {
                  cache.getTransactionManager().begin();
                  cache.put(key1, "foo");
                  cache.put(key2, "foo");
                  startSignal.await();
                  cache.getTransactionManager().commit();
               }
               catch (Exception e)
               {
                  System.out.println(this + " error " + e.getMessage());
                  e.printStackTrace();
                  exception.set(e);
               }
               finally
               {
                  endSignal.countDown();
               }
            }
         };
         t.start();
      }
      for (int i = 0; i < writersPerCategory; i++)
      {
         Thread t = new Thread("Writer Type 1 Index " + i)
         {
            public void run()
            {
               try
               {
                  cache.getTransactionManager().begin();
                  cache.put(key3, "foo");
                  cache.put(key2, "foo");
                  startSignal.await();
                  cache.getTransactionManager().commit();
               }
               catch (Exception e)
               {
                  System.out.println(this + " error " + e.getMessage());
                  e.printStackTrace();
                  exception.set(e);
               }
               finally
               {
                  endSignal.countDown();
               }
            }
         };
         t.start();
      }

      long time = System.currentTimeMillis();
      startSignal.countDown();
      endSignal.await();
      System.out.println("Done in " + (System.currentTimeMillis() - time) + " ms");
      assertNull(exception.get());
   }

   public static class CacheKey
   {
      private final String id;

      public CacheKey(String id)
      {
         this.id = id;
      }

      @Override
      public int hashCode()
      {
         final int prime = 31;
         int result = 1;
         result = prime * result + ((id == null) ? 0 : id.hashCode());
         result = prime * result;
         return result;
      }

      @Override
      public boolean equals(Object obj)
      {
         if (this == obj)
            return true;
         if (obj == null)
            return false;
         if (getClass() != obj.getClass())
            return false;
         CacheKey other = (CacheKey)obj;
         if (id == null)
         {
            if (other.id != null)
               return false;
         }
         else if (!id.equals(other.id))
            return false;
         return true;
      }

      @Override
      public String toString()
      {
         return "CacheKey [id=" + id + "]";
      }
   }
}