package org.jboss.cache.loader;

import junit.framework.Test;
import junit.framework.TestSuite;
import org.jboss.cache.Fqn;
import org.jboss.cache.TreeCache;
import org.jboss.cache.misc.TestingUtil;
import org.jboss.cache.transaction.DummyTransactionManager;

import javax.transaction.NotSupportedException;
import javax.transaction.RollbackException;
import javax.transaction.SystemException;
import javax.transaction.Transaction;
import java.io.File;
import java.util.Set;

/**
 * @author Bela Ban
 * @version $Id: TxCacheLoaderTest.java,v 1.4 2006/05/05 12:06:58 msurtani Exp $
 */
public class TxCacheLoaderTest extends AbstractCacheLoaderTestBase {
   TreeCache cache1, cache2;
    private Fqn fqn = Fqn.fromString("/one/two/three");

    protected void setUp() throws Exception {
       super.setUp();

       String tmpLoc = System.getProperty("java.io.tmpdir", "/tmp");
       String location = tmpLoc + File.separator + "TxCacheLoaderTest1";

       cache1=new TreeCache();
       cache1.setCacheMode("repl_sync");
       cache1.setTransactionManagerLookupClass("org.jboss.cache.DummyTransactionManagerLookup");

       cache1.setCacheLoaderConfiguration(getSingleCacheLoaderConfig("", "org.jboss.cache.loader.FileCacheLoader", "location=" + location, false, false, false));
       // cache1.setReplQueueInterval(3000);
       cache1.createService();
       cache1.startService();

       location = tmpLoc + File.separator + "TxCacheLoaderTest2";

       cache2=new TreeCache();
       cache2.setCacheMode("repl_sync");
       cache2.setTransactionManagerLookupClass("org.jboss.cache.DummyTransactionManagerLookup");
       cache2.setCacheLoaderConfiguration(getSingleCacheLoaderConfig("", "org.jboss.cache.loader.FileCacheLoader", "location=" + location, false, false, false));
       cache2.setLockAcquisitionTimeout(2000);
       // cache2.setReplQueueInterval(3000);
       cache2.createService();
       cache2.startService();
    }

   protected void tearDown() throws Exception {
      super.tearDown();

      // clean up cache loaders!!
      cache1.remove(Fqn.ROOT);

      cache1.stopService();
      cache1.destroyService();
      cache2.stopService();
      cache2.destroyService();
   }





   public void testTxPutCommit() throws Exception, NotSupportedException {
      DummyTransactionManager mgr=DummyTransactionManager.getInstance();
      mgr.begin();
      Transaction tx=mgr.getTransaction();


      cache1.put(fqn, "key1", "val1");
      cache1.put("/one/two/three/four", "key2", "val2");
      assertNull(cache2.get(fqn, "key1"));
      assertNull(cache2.get("/one/two/three/four", "key2"));
      tx.commit();
      assertNotNull(cache1.getKeys(fqn));
      Set children=cache1.getChildrenNames("/one");
      assertEquals(1, children.size());
      TestingUtil.sleepThread(2000);
      assertEquals("val1", cache2.get(fqn, "key1"));
      assertEquals("val2", cache2.get("/one/two/three/four", "key2"));
   }



   public void testTxPrepareAndRollback() throws Exception, NotSupportedException {
      final DummyTransactionManager mgr=DummyTransactionManager.getInstance();
      mgr.begin();
      Transaction tx1=mgr.getTransaction();

      cache1.setLockAcquisitionTimeout(1500);
      cache2.setLockAcquisitionTimeout(1500);


      Thread locker=new Thread() {
         Transaction tx2=null;
         public void run() {
            try {
               mgr.begin();
               tx2=mgr.getTransaction();
               cache2.put(fqn, "block-key1", "block-val1"); // acquires a lock on cache2./one/two/three
               TestingUtil.sleepThread(5000);
            }
            catch(Exception e) {
               e.printStackTrace();
            }
            finally {
               if(tx2 != null) {
                  try {
                     tx2.rollback();
                  }
                  catch(SystemException e) {
                     e.printStackTrace();
                  }
               }
            }
         }
      };

      locker.start();
      TestingUtil.sleepThread(1000);

      cache1.put(fqn, "key1", "val1");
      cache1.put("/one/two/three/four", "key2", "val2");

      try {
         tx1.commit(); // prepare() on cache2 will fail due to lock held by locker thread
         fail("commit() should fail because we cannot acquire the lock on cache2");
      }
      catch(RollbackException rollback) {
         System.out.println("--- TX was rolled back (as expected)");
         assertTrue(true);
      }

      assertNull(cache1.get(fqn, "key1"));
      assertNull(cache1.get("/one/two/three/four", "key1"));

   }


   public void testPutAfterTxCommit() throws Exception, NotSupportedException
   {
      DummyTransactionManager mgr=DummyTransactionManager.getInstance();
      mgr.begin();
      Transaction tx=mgr.getTransaction();

      cache1.put(fqn, "key1", "val1");
      assertTrue(cache1.exists(fqn));
      tx.commit();
      assertTrue(cache1.exists(fqn));
      cache1.put("/a/b/c", null); // should be run outside a TX !
      assertTrue(cache1.exists("/a/b/c"));
   }

   public void testPutAfterTxRollback() throws Exception, NotSupportedException
   {
      DummyTransactionManager mgr=DummyTransactionManager.getInstance();
      mgr.begin();
      Transaction tx=mgr.getTransaction();

      cache1.put(fqn, "key1", "val1");
      assertTrue(cache1.exists(fqn));
      tx.rollback();
      assertFalse(cache1.getCacheLoader().exists(fqn));
      assertFalse(cache1.exists(fqn));
      cache1.put("/a/b/c", null); // should be run outside a TX !
      assertTrue(cache1.exists("/a/b/c"));
   }

   public static Test suite() {
      return new TestSuite(TxCacheLoaderTest.class);
   }

   public static void main(String[] args) {
      junit.textui.TestRunner.run(suite());
   }

}
