/*
* JBoss, the OpenSource J2EE webOS
*
* Distributable under LGPL license.
* See terms of license at gnu.org.
*/
package org.jboss.test.cache.test.local;

import junit.framework.Test;
import junit.framework.TestCase;
import junit.framework.TestSuite;
import org.jboss.cache.CacheException;
import org.jboss.cache.PropertyConfigurator;
import org.jboss.cache.TreeCache;
import org.jboss.cache.lock.IsolationLevel;
import org.jboss.cache.lock.TimeoutException;
import org.jboss.cache.transaction.DummyTransactionManager;
import org.jboss.logging.Logger;

import javax.naming.Context;
import javax.naming.InitialContext;
import javax.transaction.UserTransaction;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Properties;
import java.util.Set;

/**
 *
 * Unit test for local TreeCache with concurrent transactions.
 * Uses locking and multiple threads to test concurrent r/w access to the tree.
 *
 * @author <a href="mailto:spohl@users.sourceforge.net">Stefan Pohl</a>
 * @author Ben Wang
 * @version $Revision: 1.6.4.2 $
 *
 */
public class TxConcurrentBankUnitTestCase extends TestCase {
   TreeCache cache;
   private static Logger logger_ = Logger.getLogger(TxConcurrentBankUnitTestCase.class);
   static Properties p = null;
   String old_factory = null;
   final String FACTORY = "org.jboss.cache.transaction.DummyContextFactory";
   final String NODE = "/cachetest";
   final int ROLLBACK_CHANCE = 100;

   static String customer[] = { "cu1", "cu2", "cu3" };
   static final int BOOKINGS = 1000;
   static boolean _testFailedinThread = false;

   public TxConcurrentBankUnitTestCase(String name)
   {
      super(name);
   }

   public void failMain() {
      _testFailedinThread=true;
   }

   public void setUp() throws Exception
   {
      super.setUp();
      old_factory = System.getProperty(Context.INITIAL_CONTEXT_FACTORY);
      System.setProperty(Context.INITIAL_CONTEXT_FACTORY, FACTORY);
      DummyTransactionManager.getInstance();
      if (p == null) {
         p = new Properties();
         p.put(Context.INITIAL_CONTEXT_FACTORY, "org.jboss.cache.transaction.DummyContextFactory");
      }

      cache = new TreeCache();

      PropertyConfigurator config = new PropertyConfigurator();
      config.configure(cache, "META-INF/local-eviction-service.xml");

      // XML file above only sets REPEATABLE-READ
      cache.setIsolationLevel(IsolationLevel.SERIALIZABLE);

      cache.createService();
      cache.startService();
   }

   public void tearDown() throws Exception
   {
      super.tearDown();
      cache.stopService();
      // BW. kind of a hack to destroy jndi binding and thread local tx before next run.
      DummyTransactionManager.destroy();
      if (old_factory != null) {
         System.setProperty(Context.INITIAL_CONTEXT_FACTORY, old_factory);
         old_factory = null;
      }
   }

   public void testConcurrentBooking()
   {
      Teller one, two;
      try {
         if(cache.get(NODE)==null) {
            cache.put(NODE, "cu1", new Integer(1000));
            cache.put(NODE, "cu2", new Integer(1000));
            cache.put(NODE, "cu3", new Integer(1000));
         }

         one = new Teller("one", cache);
         two = new Teller("two", cache);

         one.start();
         sleep(100);
         two.start();
         one.join();
         two.join();

         log("lock info:\n" + cache.printLockInfo()+_testFailedinThread);
         if(_testFailedinThread) fail();
      } catch (Exception e) {
         e.printStackTrace();
         fail(e.toString());
      } finally {
         /*
         try {
            cache.remove(NODE);
         } catch (Exception e) {
            e.printStackTrace();
            fail();
         }
         */
      }
   }

   void sleep(long timeout)
   {
      try {
         Thread.sleep(timeout);
      } catch (InterruptedException e) {
      }
   }

   static void log(String msg)
   {
//      System.out.println("-- [" + Thread.currentThread() + "]: " + msg);
      logger_.info("-- [" + Thread.currentThread() + "]: " + msg);
   }

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

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

   class Teller extends Thread
   {
      TreeCache cache;

      public Teller(String str, TreeCache cache)
      {
         super(str);
         this.cache = cache;
      }

      public void run()
      {
         int count = customer.length;
         UserTransaction tx = null;
         try {
            tx = (UserTransaction) new InitialContext(p).lookup("UserTransaction");

            boolean again = false;
            int src = 0;
            int dst = 0;
            int amo = 0;
            int anz =0;
            while(anz<BOOKINGS) {
               if(!again) {
                  src = (int) (Math.random()*count);
                  dst = (int) (Math.random()*(count-1));
                  amo =1+ (int) (Math.random()*20);
                  if(dst>=src) dst++;
               }

               tx.begin();
               HashMap accounts = getAccounts(); // read lock on NODE
               tx.commit();  // releases read lock

               int sum = sumAccounts(accounts);
               log(anz+": "+accounts+" Summe: "+sum);
               // the sum of all accounts always has to be 3000
               if(sum!=3000) {
                  failMain();
                  return; // terminate thread
               }
               assertEquals("the sum of all accounts always has to be 3000", 3000, sum);

               try {
                  tx.begin();
                  deposit(customer[src], customer[dst], amo, tx);  // gets write lock
                  tx.commit(); // releases write lock
                  again = false;
               }
               catch(TimeoutException timeout_ex) {
                  System.out.println("transaction is rolled back, will try again (ex=" + timeout_ex.getClass() + ")");
                  tx.rollback();
                  again = true;
               }
               catch(Throwable e) {
                  System.out.println("transaction is rolled back, will try again (ex=" + e.getMessage() + ")");
                  tx.rollback();
                  again = true;
               }
               anz++;
               yield();
            }
         } catch (Throwable t) {
            t.printStackTrace();
            fail(t.toString());
         }
      }
      /**
       * Posting
       */
      public void deposit(String from, String to, int amount, UserTransaction tx) throws Exception {
         log("deposit("+from+", "+to+", "+amount+") called.");
         int act;
         // debit
         act = ((Integer) cache.get(NODE, from)).intValue();
         cache.put(NODE, from, new Integer(act-amount));
         log("deposit("+from+", "+to+", "+amount+") debited.");

         // eventually rollback the transaction
         if((int) (Math.random()*ROLLBACK_CHANCE) == 0) {
            log("!!!manually set rollback ("+from+", "+to+", "+amount+").");
            tx.setRollbackOnly();
            throw new Exception("Manually set rollback!");
         }

         // credit
         act = ((Integer) cache.get(NODE, to)).intValue();
         cache.put(NODE, to, new Integer(act+amount));

         log("deposit("+from+", "+to+", "+amount+") finished.");
      }
      /**
       * retrieving amounts of accounts
       */
      public HashMap getAccounts() throws CacheException {
         log("getAccounts() called.");
         HashMap result = new HashMap();
         try {
            Set set = cache.getKeys(NODE);  // gets read lock
            Iterator iter = set.iterator();
            while(iter.hasNext()) {
               String name = (String) iter.next();
               result.put(name, cache.get(NODE, name));
            }
            return result;
         } catch(CacheException ce) {
            throw ce;
         }
      }
      protected int sumAccounts(HashMap map) {
         Iterator iter = map.values().iterator();
         int result = 0;
         while(iter.hasNext()) {
            result += ((Integer) iter.next()).intValue();
         }
         return result;
      }
   }
}