/*
 * Copyright 2011 Red Hat, Inc. and/or its affiliates.
 *
 * This is free software; you can redistribute it and/or modify it
 * under the terms of the GNU Lesser General Public License as
 * published by the Free Software Foundation; either version 2.1 of
 * the License, or (at your option) any later version.
 *
 * This software is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public
 * License along with this library; if not, write to the Free Software
 * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
 * 02110-1301 USA
 */

package org.infinispan;

import org.infinispan.config.Configuration;
import org.infinispan.context.Flag;
import org.infinispan.manager.EmbeddedCacheManager;
import org.infinispan.test.SingleCacheManagerTest;
import org.infinispan.test.fwk.TestCacheManagerFactory;
import org.infinispan.transaction.LockingMode;
import org.infinispan.transaction.lookup.JBossStandaloneJTAManagerLookup;
import org.infinispan.util.concurrent.IsolationLevel;
import org.testng.annotations.Test;

import javax.transaction.Status;
import javax.transaction.TransactionManager;
import java.util.concurrent.Callable;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

/**
 * // TODO: Document this
 *
 * @author Galder Zamarreño
 * @since // TODO
 */
@Test(groups = "functional", testName = "SampleTest")
public class Sample2Test extends SingleCacheManagerTest {

   @Override
   protected EmbeddedCacheManager createCacheManager() throws Exception {
      Configuration cfg = new Configuration().fluent()
            .clustering().mode(Configuration.CacheMode.REPL_SYNC)
            .locking()
            .concurrencyLevel(10000).isolationLevel(IsolationLevel.REPEATABLE_READ)
            .lockAcquisitionTimeout(100L).useLockStriping(false).writeSkewCheck(true)
            .transaction()
            .lockingMode(LockingMode.PESSIMISTIC)
            .recovery()
            .transactionManagerLookup(new JBossStandaloneJTAManagerLookup())
            .build();
      return TestCacheManagerFactory.createCacheManager(cfg);
   }

   public void test000() throws Exception {
      final Cache<Integer, String> cache = cacheManager.getCache();
      final TransactionManager tm = cache.getAdvancedCache().getTransactionManager();
      final ExecutorService e = Executors.newCachedThreadPool();
      final CountDownLatch waitLatch = new CountDownLatch(1);
      final CountDownLatch continueLatch = new CountDownLatch(1);
      cache.put(1, "v1");

      Future<Void> f1 = e.submit(new Callable<Void>() {
         @Override
         public Void call() throws Exception {
            tm.begin();
            try {
               cache.put(1, "v2");
               waitLatch.countDown();
               continueLatch.await();
            } catch (Exception e) {
               tm.setRollbackOnly();
               throw e;
            } finally {
               if (tm.getStatus() == Status.STATUS_ACTIVE)
                  tm.commit();
               else
                  tm.rollback();
            }
            return null;
         }
      });


      Future<Void> f2 = e.submit(new Callable<Void>() {
         @Override
         public Void call() throws Exception {
            waitLatch.await();
            tm.begin();
            try {
               AdvancedCache<Integer, String> silentCache = cache.getAdvancedCache().withFlags(
                  Flag.FAIL_SILENTLY, Flag.ZERO_LOCK_ACQUISITION_TIMEOUT);
               silentCache.put(1, "v3");
               assert !silentCache.lock(1);
               assert "v2".equals(cache.get(1));
            } catch (Exception e) {
               tm.setRollbackOnly();
               throw e;
            } finally {
               if (tm.getStatus() == Status.STATUS_ACTIVE)
                  tm.commit();
               else
                  tm.rollback();

               continueLatch.countDown();
            }
            return null;
         }
      });

      f1.get();
      f2.get();
   }

}