package org.jboss.cache.optimistic;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.jboss.cache.*;
import org.jboss.cache.misc.TestingUtil;
import org.jboss.cache.loader.SamplePojo;
import org.jboss.cache.transaction.DummyTransactionManager;

import javax.transaction.Transaction;

/**
 * @author xenephon
 */
public class AsyncFullStackInterceptorTest extends AbstractOptimisticTestCase
{

    private Log log = LogFactory.getLog(AsyncFullStackInterceptorTest.class);

    /**
     * @param name
     */
    public AsyncFullStackInterceptorTest(String name)
    {
        super(name);
    }

    private int groupIncreaser = 0;


    public void testSingleInstanceRollback() throws Exception
    {
        groupIncreaser++;
        TreeCache cache = createAsyncReplicatedCache();

        DummyTransactionManager mgr = DummyTransactionManager.getInstance();
        assertNull(mgr.getTransaction());

        mgr.begin();

        assertEquals(0, cache.getTransactionTable().getNumGlobalTransactions());
        assertEquals(0, cache.getTransactionTable().getNumLocalTransactions());

        SamplePojo pojo = new SamplePojo(21, "test");

        cache.put("/one/two", "key1", pojo);

        mgr.rollback();

        assertNull(mgr.getTransaction());
        assertEquals(0, cache.getTransactionTable().getNumGlobalTransactions());
        assertEquals(0, cache.getTransactionTable().getNumLocalTransactions());

        assertEquals(false, cache.exists(Fqn.fromString("/one/two")));
        assertNull(cache.get(Fqn.fromString("/")).getChild("one"));

        destroyCache(cache);

    }

    public void testSingleInstanceDuplicateCommit() throws Exception
    {
        groupIncreaser++;
        TreeCache cache = createAsyncReplicatedCache();

        DummyTransactionManager mgr = DummyTransactionManager.getInstance();
        assertNull(mgr.getTransaction());

        mgr.begin();

        assertEquals(0, cache.getTransactionTable().getNumGlobalTransactions());
        assertEquals(0, cache.getTransactionTable().getNumLocalTransactions());

        SamplePojo pojo = new SamplePojo(21, "test");

        cache.put("/one/two", "key1", pojo);

        mgr.commit();

        assertNull(mgr.getTransaction());

        boolean fail = false;
        try
        {
            mgr.commit();
        }
        catch (Exception e)
        {
            fail = true;

        }

        assertEquals(true, fail);
        assertEquals(0, cache.getTransactionTable().getNumGlobalTransactions());
        assertEquals(0, cache.getTransactionTable().getNumLocalTransactions());

        assertTrue(cache.exists(Fqn.fromString("/one/two")));
        assertNotNull(cache.get(Fqn.fromString("/")).getChild("one"));
        assertEquals(false, cache.get(Fqn.fromString("/")).getLock()
                .isLocked());
        assertEquals(false, cache.get(Fqn.fromString("/one")).getLock()
                .isLocked());
        assertEquals(false, cache.get(Fqn.fromString("/one/two"))
                .getLock().isLocked());
        assertNotNull(cache.get(Fqn.fromString("/one")).getChild("two"));
        assertNotNull(cache.get(Fqn.fromString("/one/two"), "key1"));

        destroyCache(cache);

    }

    public void testValidationFailCommit() throws Exception
    {
        groupIncreaser++;
        TreeCache cache = createAsyncReplicatedCache();

        DummyTransactionManager mgr = DummyTransactionManager.getInstance();
        assertNull(mgr.getTransaction());

        mgr.begin();
        Transaction tx = mgr.getTransaction();
        assertEquals(0, cache.getTransactionTable().getNumGlobalTransactions());
        assertEquals(0, cache.getTransactionTable().getNumLocalTransactions());

        SamplePojo pojo = new SamplePojo(21, "test");

        cache.put("/one/two", "key1", pojo);

        mgr.suspend();

        assertEquals(1, cache.getTransactionTable().getNumGlobalTransactions());
        assertEquals(1, cache.getTransactionTable().getNumLocalTransactions());

        assertNull(mgr.getTransaction());

        mgr.begin();

        SamplePojo pojo2 = new SamplePojo(22, "test2");

        cache.put("/one/two", "key1", pojo2);

        mgr.commit();

        assertEquals(1, cache.getTransactionTable().getNumGlobalTransactions());
        assertEquals(1, cache.getTransactionTable().getNumLocalTransactions());

        mgr.resume(tx);

        boolean fail = false;
        try
        {
            mgr.commit();
        }
        catch (Exception e)
        {
            fail = true;

        }
        assertNull(mgr.getTransaction());
        assertEquals(0, cache.getTransactionTable().getNumGlobalTransactions());
        assertEquals(0, cache.getTransactionTable().getNumLocalTransactions());

        assertEquals(true, fail);

        assertTrue(cache.exists(Fqn.fromString("/one/two")));
        assertNotNull(cache.get(Fqn.fromString("/")).getChild("one"));
        assertEquals(false, cache.get(Fqn.fromString("/")).getLock()
                .isLocked());
        assertEquals(false, cache.get(Fqn.fromString("/one")).getLock()
                .isLocked());
        assertEquals(false, cache.get(Fqn.fromString("/one/two")).getLock().isLocked());
        assertNotNull(cache.get(Fqn.fromString("/one")).getChild("two"));
        assertEquals(pojo2, cache.get(Fqn.fromString("/one/two"), "key1"));

        destroyCache(cache);

    }

    public void test2InstanceCommit() throws Exception
    {
        groupIncreaser++;
        TreeCache cache = createAsyncReplicatedCache();
        TreeCache cache2 = createAsyncReplicatedCache();

        DummyTransactionManager mgr = DummyTransactionManager.getInstance();
        assertNull(mgr.getTransaction());

        mgr.begin();

        assertEquals(0, cache.getTransactionTable().getNumGlobalTransactions());
        assertEquals(0, cache.getTransactionTable().getNumLocalTransactions());

        SamplePojo pojo = new SamplePojo(21, "test");

        cache.put("/one/two", "key1", pojo);

        mgr.commit();

        // cache asserts
        assertNull(mgr.getTransaction());
        assertEquals(0, cache.getTransactionTable().getNumGlobalTransactions());
        assertEquals(0, cache.getTransactionTable().getNumLocalTransactions());

        assertTrue(cache.exists(Fqn.fromString("/one/two")));
        assertNotNull(cache.get(Fqn.fromString("/")).getChild("one"));
        assertNotNull(cache.get(Fqn.fromString("/one/two"), "key1"));

        assertTrue(cache.exists(Fqn.fromString("/one/two")));
        assertNotNull(cache.get(Fqn.fromString("/")).getChild("one"));
        assertEquals(false, cache.get(Fqn.fromString("/")).getLock()
                .isLocked());
        assertEquals(false, cache.get(Fqn.fromString("/one")).getLock()
                .isLocked());
        assertEquals(false, cache.get(Fqn.fromString("/one/two")).getLock().isLocked());
        assertNotNull(cache.get(Fqn.fromString("/one")).getChild("two"));
        assertNotNull(cache.get(Fqn.fromString("/one/two"), "key1"));

        // let async calls propagate
        TestingUtil.sleepThread((long)1000);

        // cache2 asserts
        assertEquals(0, cache2.getTransactionTable().getNumGlobalTransactions());
        assertEquals(0, cache2.getTransactionTable().getNumLocalTransactions());

        assertTrue(cache2.exists(Fqn.fromString("/one/two")));
        assertNotNull(cache2.get(Fqn.fromString("/")).getChild("one"));
        assertNotNull(cache2.get(Fqn.fromString("/one/two"), "key1"));

        assertTrue(cache2.exists(Fqn.fromString("/one/two")));
        assertNotNull(cache2.get(Fqn.fromString("/")).getChild("one"));
        assertEquals(false, cache2.get(Fqn.fromString("/")).getLock().isLocked());
        assertEquals(false, cache2.get(Fqn.fromString("/one")).getLock().isLocked());
        assertEquals(false, cache2.get(Fqn.fromString("/one/two")).getLock().isLocked());
        assertNotNull(cache2.get(Fqn.fromString("/one")).getChild("two"));
        assertNotNull(cache2.get(Fqn.fromString("/one/two"), "key1"));

        destroyCache(cache);
        destroyCache(cache2);
    }

    public void test2InstanceRemove() throws Exception
    {
        groupIncreaser++;
        TreeCache cache = createAsyncReplicatedCache();
        TreeCache cache2 = createAsyncReplicatedCache();

        DummyTransactionManager mgr = DummyTransactionManager.getInstance();
        assertNull(mgr.getTransaction());

        mgr.begin();

        assertEquals(0, cache.getTransactionTable().getNumGlobalTransactions());
        assertEquals(0, cache.getTransactionTable().getNumLocalTransactions());

        SamplePojo pojo = new SamplePojo(21, "test");

        cache.put("/one/two", "key1", pojo);

        mgr.commit();

        // cache asserts
        assertNull(mgr.getTransaction());
        assertEquals(0, cache.getTransactionTable().getNumGlobalTransactions());
        assertEquals(0, cache.getTransactionTable().getNumLocalTransactions());

        assertTrue(cache.exists(Fqn.fromString("/one/two")));
        assertNotNull(cache.get(Fqn.fromString("/")).getChild("one"));
        assertNotNull(cache.get(Fqn.fromString("/one/two"), "key1"));

        assertTrue(cache.exists(Fqn.fromString("/one/two")));
        assertNotNull(cache.get(Fqn.fromString("/")).getChild("one"));
        assertEquals(false, cache.get(Fqn.fromString("/")).getLock()
                .isLocked());
        assertEquals(false, cache.get(Fqn.fromString("/one")).getLock()
                .isLocked());
        assertEquals(false, cache.get(Fqn.fromString("/one/two"))
                .getLock().isLocked());
        assertNotNull(cache.get(Fqn.fromString("/one")).getChild("two"));
        assertNotNull(cache.get(Fqn.fromString("/one/two"), "key1"));

        // let async calls propagate
        TestingUtil.sleepThread((long)1000);

        // cache2 asserts
        assertEquals(0, cache2.getTransactionTable().getNumGlobalTransactions());
        assertEquals(0, cache2.getTransactionTable().getNumLocalTransactions());

        assertTrue(cache2.exists(Fqn.fromString("/one/two")));
        assertNotNull(cache2.get(Fqn.fromString("/")).getChild("one"));
        assertNotNull(cache2.get(Fqn.fromString("/one/two"), "key1"));

        assertTrue(cache2.exists(Fqn.fromString("/one/two")));
        assertNotNull(cache2.get(Fqn.fromString("/")).getChild("one"));
        assertEquals(false, cache2.get(Fqn.fromString("/")).getLock().isLocked());
        assertEquals(false, cache2.get(Fqn.fromString("/one")).getLock().isLocked());
        assertEquals(false, cache2.get(Fqn.fromString("/one/two")).getLock().isLocked());
        assertNotNull(cache2.get(Fqn.fromString("/one")).getChild("two"));
        assertNotNull(cache2.get(Fqn.fromString("/one/two"), "key1"));

        cache.remove("/one/two");
        log.debug("  C1 " + cache.get("/one/two"));
        assertEquals(false, cache.exists("/one/two"));
        assertEquals(null, cache.get("/one/two", "key1"));

        // let async calls propagate
        TestingUtil.sleepThread((long)1000);

        log.debug("  C2 " + cache2.get("/one/two"));
        assertEquals(false, cache2.exists("/one/two"));
        assertEquals(null, cache2.get("/one/two", "key1"));

        destroyCache(cache);
        destroyCache(cache2);
    }

    public void testValidationFailCommit2Instances() throws Exception
    {
        groupIncreaser++;
        TreeCache cache = createAsyncReplicatedCache();
        TreeCache cache2 = createAsyncReplicatedCache();

        DummyTransactionManager mgr = DummyTransactionManager.getInstance();
        assertNull(mgr.getTransaction());

        mgr.begin();
        Transaction tx = mgr.getTransaction();
        assertEquals(0, cache.getTransactionTable().getNumGlobalTransactions());
        assertEquals(0, cache.getTransactionTable().getNumLocalTransactions());

        SamplePojo pojo = new SamplePojo(21, "test");

        cache.put("/one/two", "key1", pojo);

        mgr.suspend();

        assertEquals(1, cache.getTransactionTable().getNumGlobalTransactions());
        assertEquals(1, cache.getTransactionTable().getNumLocalTransactions());

        GlobalTransaction gtx = cache.getCurrentTransaction(tx);
        TransactionTable table = cache.getTransactionTable();
        OptimisticTransactionEntry entry = (OptimisticTransactionEntry) table
                .get(gtx);


        assertEquals(3, entry.getTransactionWorkSpace().getNodes().size());
        assertNull(mgr.getTransaction());

        mgr.begin();

        SamplePojo pojo2 = new SamplePojo(22, "test2");

        cache2.put("/one/two", "key1", pojo2);

        mgr.commit();

        // let async calls propagate
        TestingUtil.sleepThread((long)1000);

        assertEquals(1, cache.getTransactionTable().getNumGlobalTransactions());
        assertEquals(1, cache.getTransactionTable().getNumLocalTransactions());

        mgr.resume(tx);

        boolean fail = false;
        try
        {
            mgr.commit();
        }
        catch (Exception e)
        {
            fail = true;

        }

        assertEquals(true, fail);
        assertNull(mgr.getTransaction());
        assertEquals(0, cache.getTransactionTable().getNumGlobalTransactions());
        assertEquals(0, cache.getTransactionTable().getNumLocalTransactions());

        assertEquals(0, entry.getTransactionWorkSpace().getNodes().size());

        assertTrue(cache.exists(Fqn.fromString("/one/two")));
        assertNotNull(cache.get(Fqn.fromString("/")).getChild("one"));
        assertEquals(false, cache.get(Fqn.fromString("/")).getLock().isLocked());
        assertEquals(false, cache.get(Fqn.fromString("/one")).getLock().isLocked());
        assertEquals(false, cache.get(Fqn.fromString("/one/two")).getLock().isLocked());
        assertNotNull(cache.get(Fqn.fromString("/one")).getChild("two"));
        assertEquals(pojo2, cache.get(Fqn.fromString("/one/two"), "key1"));

        destroyCache(cache);
        destroyCache(cache2);

    }

    protected TreeCache createAsyncReplicatedCache() throws Exception
    {
        return createReplicatedCache("temp" + groupIncreaser, TreeCache.REPL_ASYNC);
    }

}
