/*
 * JBoss, Home of Professional Open Source
 *
 * Distributable under LGPL license.
 * See terms of license at gnu.org.
 */

package org.jboss.cache.pojo.rollback;

import static org.testng.AssertJUnit.assertEquals;
import static org.testng.AssertJUnit.fail;

import java.util.ArrayList;
import java.util.List;
import java.util.Properties;

import javax.naming.Context;
import javax.naming.InitialContext;
import javax.naming.NamingException;
import javax.transaction.NotSupportedException;
import javax.transaction.RollbackException;
import javax.transaction.SystemException;
import javax.transaction.UserTransaction;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.jboss.cache.config.Configuration.CacheMode;
import org.jboss.cache.factories.UnitTestCacheConfigurationFactory;
import org.jboss.cache.util.TestingUtil;
import org.jboss.cache.pojo.PojoCache;
import org.jboss.cache.pojo.PojoCacheFactory;
import org.jboss.cache.pojo.test.Person;
import org.jboss.cache.transaction.DummyTransactionManager;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;

/**
 */

@Test(groups = {"functional"})
public class ReplicatedTxTest
{
   Log log = LogFactory.getLog(org.jboss.cache.pojo.rollback.ReplicatedTxTest.class);
   PojoCache cache, cache1;
   final String FACTORY = "org.jboss.cache.transaction.DummyContextFactory";
   DummyTransactionManager tx_mgr;
   Throwable t1_ex, t2_ex;
   long start = 0;



   @BeforeMethod(alwaysRun = true)
   protected void setUp() throws Exception
   {
      log.info("setUp() ....");
      boolean toStart = false;
      cache = PojoCacheFactory.createCache(UnitTestCacheConfigurationFactory.createConfiguration(CacheMode.REPL_SYNC), toStart);
      cache.start();
      cache1 = PojoCacheFactory.createCache(UnitTestCacheConfigurationFactory.createConfiguration(CacheMode.REPL_SYNC), toStart);
      cache1.start();

      System.setProperty(Context.INITIAL_CONTEXT_FACTORY, FACTORY);

      tx_mgr = DummyTransactionManager.getInstance();
      t1_ex = t2_ex = null;
   }

   @AfterMethod(alwaysRun = true)
   protected void tearDown() throws Exception
   {
      cache.stop();
      cache1.stop();

      DummyTransactionManager.destroy();
   }

//   public void testDummy() {}

   UserTransaction getTransaction() throws SystemException, NotSupportedException, NamingException
   {
      Properties prop = new Properties();
      prop.put(Context.INITIAL_CONTEXT_FACTORY,
               "org.jboss.cache.transaction.DummyContextFactory");
      return (UserTransaction) new InitialContext(prop).lookup("UserTransaction");
   }

   private Person createPerson(String id, String name, int age)
   {
      Person p = new Person();
      p.setName(name);
      p.setAge(age);
      return p;
   }

   public void testSimple() throws Exception
   {
      log.info("testSimple() ....");
      UserTransaction tx = getTransaction();
      tx.begin();
      Person p = createPerson("/person/test1", "Harald Gliebe", 32);
      cache.attach("/person/test1", p);

      tx.commit();
      tx.begin();
      p.setName("Benoit");
      tx.commit();
      Person p1 = (Person) cache1.find("/person/test1");
      assertEquals("Benoit", p.getName());
      assertEquals("Benoit", p1.getName());
      tx.begin();
      p1.setAge(61);
      tx.commit();
      assertEquals(61, p.getAge());
      assertEquals(61, p1.getAge());
   }

   /**
    * Concurrent puts (whole POJO) from the same cache instance (different threads) with rollback.
    */
   public void testConcurrentPuts() throws Exception
   {
      Thread t1 = new Thread()
      {
         public void run()
         {
            try
            {
               Person p = createPerson("/person/test6", "p6", 50);
               List<String> lang = new ArrayList<String>();
               lang.add("German");
               p.setLanguages(lang);
               UserTransaction tx = getTransaction();
               tx.begin();
               cache.attach("/person/test6", p);
               TestingUtil.sleepThread(17000);
               tx.commit();
            }
            catch (RollbackException rollback)
            {
               ;
            }
            catch (Exception ex)
            {
               t1_ex = ex;
            }
         }
      };

      Thread t2 = new Thread()
      {
         public void run()
         {
            UserTransaction tx = null;
            Person p = createPerson("/person/test6", "p6", 50);
            try
            {
               TestingUtil.sleepThread(1000); // give Thread1 time to createPerson
               List<String> lang = new ArrayList<String>();
               lang.add("German");
               p.setLanguages(lang);
               tx = getTransaction();
               tx.begin();
               cache.attach("/person/test6", p);
               tx.commit();
            }
            catch (RollbackException rollback)
            {
               ;
            }
            catch (Exception ex)
            {
               try
               {
                  tx.rollback();
               }
               catch (SystemException e)
               {
                  e.printStackTrace();
                  t2_ex = e;
               }
            }

            cache.attach("/person/test6", p);

         }
      };

      t1.start();
      t2.start();

      t1.join();
      t2.join();

      // t2 should rollback due to timeout while t2 should succeed
      if (t2_ex != null)
         fail("Thread1 failed: " + t2_ex);
      if (t1_ex != null)
         fail("Thread2 failed: " + t1_ex);

      int size = ((Person) cache.find("/person/test6")).getLanguages().size();
      assertEquals("number of languages", 1, size);
      size = ((Person) cache1.find("/person/test6")).getLanguages().size();
      assertEquals("number of languages", 1, size);
   }

   /**
    * Concurrent puts from the same cache instance (different threads) with rollback.
    */
   public void testConcurrentPuts1() throws Exception
   {
      Thread t1 = new Thread()
      {
         public void run()
         {
            try
            {
               List<String> lang = ((Person) cache.find("/person/test6")).getLanguages();
               UserTransaction tx = getTransaction();
               tx.begin();
               lang.add("German");
               TestingUtil.sleepThread(17000);
               tx.commit();
            }
            catch (RollbackException rollback)
            {
               ;
            }
            catch (Exception ex)
            {
               t1_ex = ex;
            }
         }
      };

      Thread t2 = new Thread()
      {
         public void run()
         {
            UserTransaction tx = null;
            try
            {
               TestingUtil.sleepThread(1000); // give Thread1 time to createPerson
               List<String> lang = ((Person) cache1.find("/person/test6")).getLanguages();
               System.out.println(lang == null);
               tx = getTransaction();
               tx.begin();
               lang.add("English");
               tx.commit();
               System.out.println("Successful!");
            }
            catch (RollbackException rollback)
            {
            }
            catch (Exception ex)
            {
               ex.printStackTrace();
               try
               {
                  tx.rollback();
               }
               catch (SystemException e)
               {
                  e.printStackTrace();
                  t2_ex = e;
               }
            }
         }
      };

      Person p = createPerson("/person/test6", "p6", 50);
      cache.attach("/person/test6", p);
      List<String> lang = new ArrayList<String>();
      lang.add("German");
      p.setLanguages(lang);

      t1.start();
      t2.start();

      t1.join();
      t2.join();

      // t2 should rollback due to timeout while t2 should succeed
      if (t2_ex != null)
         fail("Thread1 failed: " + t2_ex);
      if (t1_ex != null)
         fail("Thread2 failed: " + t1_ex);

      int size = ((Person) cache.find("/person/test6")).getLanguages().size();
      assertEquals("number of languages", 2, size);
      size = ((Person) cache1.find("/person/test6")).getLanguages().size();
      assertEquals("number of languages", 2, size);
   }

   /**
    * Concurrent puts from the different cache instances (different threads) with rollback.
    */
   public void testConcurrentPuts2() throws Exception
   {
      Thread t1 = new Thread()
      {
         public void run()
         {
            try
            {
               List<String> lang = ((Person) cache.find("/person/test6")).getLanguages();
               UserTransaction tx = getTransaction();
               tx.begin();
               lang.add("German");
               TestingUtil.sleepThread(17000);
               tx.commit();
            }
            catch (RollbackException rollback)
            {
               ;
            }
            catch (Exception ex)
            {
               t1_ex = ex;
            }
         }
      };

      Thread t2 = new Thread()
      {
         public void run()
         {
            UserTransaction tx = null;
            try
            {
               TestingUtil.sleepThread(1000); // give Thread1 time to createPerson
               List<String> lang = ((Person) cache.find("/person/test6")).getLanguages();
               tx = getTransaction();
               tx.begin();
               lang.add("English");
               tx.commit();
            }
            catch (RollbackException rollback)
            {
               ;
            }
            catch (Exception ex)
            {
               try
               {
                  tx.rollback();
               }
               catch (SystemException e)
               {
                  e.printStackTrace();
                  t2_ex = e;
               }
            }
         }
      };

      Person p = createPerson("/person/test6", "p6", 50);
      cache.attach("/person/test6", p);
      List<String> lang = new ArrayList<String>();
      lang.add("German");
      p.setLanguages(lang);

      t1.start();
      t2.start();

      t1.join();
      t2.join();

      // t2 should rollback due to timeout while t2 should succeed
      if (t2_ex != null)
         fail("Thread1 failed: " + t2_ex);
      if (t1_ex != null)
         fail("Thread2 failed: " + t1_ex);

      int size = ((Person) cache.find("/person/test6")).getLanguages().size();
      assertEquals("number of languages", 2, size);
      size = ((Person) cache1.find("/person/test6")).getLanguages().size();
      assertEquals("number of languages", 2, size);
   }

   void log(String s)
   {
      long now;
      if (start == 0)
         start = System.currentTimeMillis();
      now = System.currentTimeMillis();

      System.out.println("[" + Thread.currentThread().getName() + "] [" + (now - start) + "] " + s);
   }





}
