package org.infinispan.replication;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import javax.transaction.TransactionManager;

import org.infinispan.manager.EmbeddedCacheManager;
import org.infinispan.test.SingleCacheManagerTest;
import org.infinispan.test.fwk.TestCacheManagerFactory;
import org.testng.annotations.Test;

public class DeadlockLockingTest extends SingleCacheManagerTest {

   @Test
   public void test() throws Exception {

      ExecutorService executorService = Executors.newFixedThreadPool(2);
      final TransactionManager txnMgr = cache.getAdvancedCache().getTransactionManager();
      assert txnMgr != null;
      final AtomicInteger counter = new AtomicInteger(0);
      for (int i = 0; i < 100000; ++i) {
         final String valueStr = "V" + i;
         executorService.execute(new Runnable() {
            @Override
            public void run() {
               try {
                  txnMgr.begin();
                  cache.getAdvancedCache().lock("K1");
                  cache.put("K1", valueStr);
                  txnMgr.commit();
               } catch (Exception e) {
                  counter.incrementAndGet();
                  System.out.println("Exception is " + e);
                  try {
                     txnMgr.rollback();
                  } catch (Exception ex) {
                  }
               }
            }
         });
      }

      executorService.shutdown();
      assert executorService.awaitTermination(5000, TimeUnit.SECONDS);
      assert counter.get() == 0;
   }

   @Override
   protected EmbeddedCacheManager createCacheManager() throws Exception {
      EmbeddedCacheManager cm = TestCacheManagerFactory.createLocalCacheManager(true);
      cache = cm.getCache();
      return cm;
   }
}