/*
 * Created on 17-Feb-2005
 *
 *  
 *
 */
package org.jboss.cache.optimistic;

import org.jboss.cache.GlobalTransaction;
import org.jboss.cache.OptimisticTransactionEntry;
import org.jboss.cache.TransactionTable;
import org.jboss.cache.TreeCache;
import org.jboss.cache.interceptors.*;
import org.jboss.cache.loader.SamplePojo;
import org.jboss.cache.marshall.MethodCallFactory;
import org.jboss.cache.marshall.MethodDeclarations;
import org.jboss.cache.transaction.DummyTransactionManager;
import org.jgroups.Address;
import org.jgroups.blocks.MethodCall;

import javax.transaction.Transaction;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.ObjectInput;
import java.io.ObjectOutput;
import java.util.List;

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


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

    }

    public void testNoTransaction() throws Exception
    {

        TreeCache cache = createCache();

        Interceptor txInterceptor = new TxInterceptor();
        txInterceptor.setCache(cache);
        Interceptor createInterceptor = new OptimisticCreateIfNotExistsInterceptor();
        createInterceptor.setCache(cache);
        Interceptor nodeInterceptor = new OptimisticNodeInterceptor();
        nodeInterceptor.setCache(cache);
        MockInterceptor dummy = new MockInterceptor();
        dummy.setCache(cache);

        txInterceptor.setNext(createInterceptor);
        createInterceptor.setNext(nodeInterceptor);
        nodeInterceptor.setNext(dummy);

        cache.setInterceptorChain(txInterceptor);
        DummyTransactionManager mgr = DummyTransactionManager.getInstance();
        assertNull(mgr.getTransaction());
        assertEquals(0, cache.getTransactionTable().getNumGlobalTransactions());
        assertEquals(0, cache.getTransactionTable().getNumLocalTransactions());

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

        cache.put("/one/two", "key1", pojo);
        assertNull(mgr.getTransaction());
        assertEquals(0, cache.getTransactionTable().getNumGlobalTransactions());
        assertEquals(0, cache.getTransactionTable().getNumLocalTransactions());

        //make sure all calls were done in right order

        List calls = dummy.getAllCalled();

        assertEquals(MethodDeclarations.optimisticPrepareMethod, calls.get(0));
        assertEquals(MethodDeclarations.commitMethod, calls.get(1));
        //flesh this out a bit more

    }

    public void testLocalTransactionExists() throws Exception
    {

        TreeCache cache = createCache();
        Interceptor txInterceptor = new TxInterceptor();
        txInterceptor.setCache(cache);
        Interceptor createInterceptor = new OptimisticCreateIfNotExistsInterceptor();
        createInterceptor.setCache(cache);
        Interceptor nodeInterceptor = new OptimisticNodeInterceptor();
        nodeInterceptor.setCache(cache);
        MockInterceptor dummy = new MockInterceptor();
        dummy.setCache(cache);

        txInterceptor.setNext(createInterceptor);
        createInterceptor.setNext(nodeInterceptor);
        nodeInterceptor.setNext(dummy);

        cache.setInterceptorChain(txInterceptor);

        DummyTransactionManager mgr = DummyTransactionManager.getInstance();

        //start local transaction
        mgr.begin();
        Transaction tx = mgr.getTransaction();

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

        assertNotNull(mgr.getTransaction());
        TransactionTable txTable = cache.getTransactionTable();
        assertNull(txTable.get(tx));

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

        assertNotNull(mgr.getTransaction());
        mgr.commit();

        assertNull(mgr.getTransaction());

        List calls = dummy.getAllCalled();
        assertEquals(MethodDeclarations.optimisticPrepareMethod, calls.get(0));
        assertEquals(MethodDeclarations.commitMethod, calls.get(1));

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

    }

    public void testRollbackTransaction() throws Exception
    {

        TreeCache cache = createCache();
        Interceptor txInterceptor = new TxInterceptor();
        txInterceptor.setCache(cache);
        Interceptor createInterceptor = new OptimisticCreateIfNotExistsInterceptor();
        createInterceptor.setCache(cache);
        Interceptor nodeInterceptor = new OptimisticNodeInterceptor();
        nodeInterceptor.setCache(cache);
        MockInterceptor dummy = new MockInterceptor();
        dummy.setCache(cache);

        txInterceptor.setNext(createInterceptor);
        createInterceptor.setNext(nodeInterceptor);
        nodeInterceptor.setNext(dummy);

        cache.setInterceptorChain(txInterceptor);

        DummyTransactionManager mgr = DummyTransactionManager.getInstance();

        //start local transaction
        mgr.begin();

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

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

        assertNotNull(mgr.getTransaction());
        mgr.rollback();

        assertNull(mgr.getTransaction());

        List calls = dummy.getAllCalled();
        assertEquals(1, calls.size());
        assertEquals(MethodDeclarations.rollbackMethod, calls.get(0));


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

    }

    public void testEmptyLocalTransaction() throws Exception
    {
        TreeCache cache = createCache();
        Interceptor txInterceptor = new TxInterceptor();
        txInterceptor.setCache(cache);
        Interceptor createInterceptor = new OptimisticCreateIfNotExistsInterceptor();
        createInterceptor.setCache(cache);
        Interceptor nodeInterceptor = new OptimisticNodeInterceptor();
        nodeInterceptor.setCache(cache);
        MockInterceptor dummy = new MockInterceptor();
        dummy.setCache(cache);

        txInterceptor.setNext(createInterceptor);
        createInterceptor.setNext(nodeInterceptor);
        nodeInterceptor.setNext(dummy);

        cache.setInterceptorChain(txInterceptor);

        DummyTransactionManager mgr = DummyTransactionManager.getInstance();

        //start local transaction
        mgr.begin();

        assertNotNull(mgr.getTransaction());
        mgr.commit();

        assertNull(mgr.getTransaction());

        List calls = dummy.getAllCalled();
        assertEquals(0, calls.size());

        assertNull(mgr.getTransaction());

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

        cache.stopService();
    }

    public void testEmptyRollbackLocalTransaction() throws Exception
    {

        TreeCache cache = createCache();
        Interceptor txInterceptor = new TxInterceptor();
        txInterceptor.setCache(cache);
        Interceptor createInterceptor = new OptimisticCreateIfNotExistsInterceptor();
        createInterceptor.setCache(cache);
        Interceptor nodeInterceptor = new OptimisticNodeInterceptor();
        nodeInterceptor.setCache(cache);
        MockInterceptor dummy = new MockInterceptor();
        dummy.setCache(cache);

        txInterceptor.setNext(createInterceptor);
        createInterceptor.setNext(nodeInterceptor);
        nodeInterceptor.setNext(dummy);

        cache.setInterceptorChain(txInterceptor);

        DummyTransactionManager mgr = DummyTransactionManager.getInstance();

        //start local transaction
        mgr.begin();

        assertNotNull(mgr.getTransaction());
        mgr.rollback();

        assertNull(mgr.getTransaction());

        List calls = dummy.getAllCalled();
        assertEquals(0, calls.size());

        assertNull(mgr.getTransaction());

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

    }

    public void testLocalRollbackAftercommitTransaction() throws Exception
    {

        TreeCache cache = createCache();
        Interceptor txInterceptor = new TxInterceptor();
        txInterceptor.setCache(cache);
        Interceptor createInterceptor = new OptimisticCreateIfNotExistsInterceptor();
        createInterceptor.setCache(cache);
        Interceptor nodeInterceptor = new OptimisticNodeInterceptor();
        nodeInterceptor.setCache(cache);
        MockInterceptor dummy = new MockInterceptor();
        dummy.setCache(cache);

        txInterceptor.setNext(createInterceptor);
        createInterceptor.setNext(nodeInterceptor);
        nodeInterceptor.setNext(dummy);

        cache.setInterceptorChain(txInterceptor);

        DummyTransactionManager mgr = DummyTransactionManager.getInstance();

        //start local transaction
        mgr.begin();

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

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

        assertNotNull(mgr.getTransaction());
        mgr.commit();

        assertNull(mgr.getTransaction());

        List calls = dummy.getAllCalled();
        assertEquals(MethodDeclarations.optimisticPrepareMethod, calls.get(0));
        assertEquals(MethodDeclarations.commitMethod, calls.get(1));
        boolean failed = false;
        try
        {
            mgr.rollback();
            fail();
        }
        catch (Exception e)
        {
            e.printStackTrace();
            failed = true;
            assertTrue(true);
        }
        assertTrue(failed);
        assertNull(mgr.getTransaction());
        assertEquals(0, cache.getTransactionTable().getNumGlobalTransactions());
        assertEquals(0, cache.getTransactionTable().getNumLocalTransactions());
        cache.stopService();
    }


    public void testgtxTransactionExists() throws Exception
    {
        TreeCache cache = createCache();
        Interceptor txInterceptor = new TxInterceptor();
        txInterceptor.setCache(cache);
        Interceptor createInterceptor = new OptimisticCreateIfNotExistsInterceptor();
        createInterceptor.setCache(cache);
        Interceptor nodeInterceptor = new OptimisticNodeInterceptor();
        nodeInterceptor.setCache(cache);
        MockInterceptor dummy = new MockInterceptor();
        dummy.setCache(cache);

        txInterceptor.setNext(createInterceptor);
        createInterceptor.setNext(nodeInterceptor);
        nodeInterceptor.setNext(dummy);

        cache.setInterceptorChain(txInterceptor);

        DummyTransactionManager mgr = DummyTransactionManager.getInstance();

        //start local transaction
        mgr.begin();
        Transaction tx = mgr.getTransaction();

        //this sets
        cache.getCurrentTransaction(tx);

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

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

        assertNotNull(mgr.getTransaction());
        mgr.commit();

        assertNull(mgr.getTransaction());

        List calls = dummy.getAllCalled();
        assertEquals(MethodDeclarations.optimisticPrepareMethod, calls.get(0));
        assertEquals(MethodDeclarations.commitMethod, calls.get(1));


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

        cache.stopService();
    }


    public void testRemotePrepareTransaction() throws Exception
    {

        TreeCache cache = createCache();
        Interceptor txInterceptor = new TxInterceptor();
        txInterceptor.setCache(cache);
        Interceptor createInterceptor = new OptimisticCreateIfNotExistsInterceptor();
        createInterceptor.setCache(cache);
        Interceptor nodeInterceptor = new OptimisticNodeInterceptor();
        nodeInterceptor.setCache(cache);
        MockInterceptor dummy = new MockInterceptor();
        dummy.setCache(cache);

        txInterceptor.setNext(createInterceptor);
        createInterceptor.setNext(nodeInterceptor);
        nodeInterceptor.setNext(dummy);

        cache.setInterceptorChain(txInterceptor);

        DummyTransactionManager mgr = DummyTransactionManager.getInstance();

        //start local transaction
        mgr.begin();
        Transaction tx = mgr.getTransaction();


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

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

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

        //test local calls
        List calls = dummy.getAllCalled();
        assertEquals(MethodDeclarations.optimisticPrepareMethod, calls.get(0));
        assertEquals(MethodDeclarations.commitMethod, calls.get(1));
        assertNull(mgr.getTransaction());

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


        GlobalTransaction remoteGtx = new GlobalTransaction();

        remoteGtx.setAddress(new Address()
        {
            public boolean isMulticastAddress()
            {

                return false;
            }

            public void readExternal(ObjectInput arg0)
            {
            }

            public void writeExternal(ObjectOutput arg0)
            {
            }

            public int compareTo(Object arg0)
            {

                return 0;
            }

            public int size()
            {

                return 0;
            }

            public void writeTo(DataOutputStream arg0)
            {
            }

            public void readFrom(DataInputStream arg0)
            {
            }
        });
        //hack the method call to make it have the remote gtx
        MethodCall meth = (MethodCall) entry.getModifications().get(0);

        meth.getArgs()[0] = remoteGtx;
        //call our remote method
        MethodCall prepareMethod = MethodCallFactory.create(MethodDeclarations.optimisticPrepareMethod, new Object[]{remoteGtx, entry.getModifications(), null, remoteGtx.getAddress(), Boolean.FALSE});
        try
        {
            cache._replicate(prepareMethod);
        }
        catch (Throwable t)
        {
            fail();
        }

        //our thread should still be null
        assertNull(mgr.getTransaction());

        //there should be a registration for the remote gtx
        assertNotNull(table.get(remoteGtx));
        assertNotNull(table.getLocalTransaction(remoteGtx));

        //assert that the method has been passed up the stack
        calls = dummy.getAllCalled();
        assertEquals(MethodDeclarations.optimisticPrepareMethod, calls.get(2));

        //assert we have the tx in th table
        assertEquals(1, cache.getTransactionTable().getNumGlobalTransactions());
        assertEquals(1, cache.getTransactionTable().getNumLocalTransactions());
        cache.stopService();
    }

    public void testRemotePrepareSuspendTransaction() throws Exception
    {

        TreeCache cache = createCache();
        Interceptor txInterceptor = new TxInterceptor();
        txInterceptor.setCache(cache);
        Interceptor createInterceptor = new OptimisticCreateIfNotExistsInterceptor();
        createInterceptor.setCache(cache);
        Interceptor nodeInterceptor = new OptimisticNodeInterceptor();
        nodeInterceptor.setCache(cache);
        MockInterceptor dummy = new MockInterceptor();
        dummy.setCache(cache);

        txInterceptor.setNext(createInterceptor);
        createInterceptor.setNext(nodeInterceptor);
        nodeInterceptor.setNext(dummy);

        cache.setInterceptorChain(txInterceptor);

        DummyTransactionManager mgr = DummyTransactionManager.getInstance();

        //start local transaction
        mgr.begin();
        Transaction tx = mgr.getTransaction();


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

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

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

        //now send the remote prepare

        GlobalTransaction remoteGtx = new GlobalTransaction();

        remoteGtx.setAddress(new Address()
        {
            public boolean isMulticastAddress()
            {

                return false;
            }

            public void readExternal(ObjectInput arg0)
            {
            }

            public void writeExternal(ObjectOutput arg0)
            {
            }

            public int compareTo(Object arg0)
            {

                return 0;
            }

            public int size()
            {

                return 0;
            }

            public void writeTo(DataOutputStream arg0)
            {
            }

            public void readFrom(DataInputStream arg0)
            {
            }
        });
        //hack the method call to make it have the remote gtx
        MethodCall meth = (MethodCall) entry.getModifications().get(0);

        meth.getArgs()[0] = remoteGtx;
        //call our remote method
        MethodCall prepareMethod = MethodCallFactory.create(MethodDeclarations.optimisticPrepareMethod, new Object[]{remoteGtx, entry.getModifications(), null, remoteGtx.getAddress(), Boolean.FALSE});
        try
        {
            cache._replicate(prepareMethod);
        }
        catch (Throwable t)
        {
            t.printStackTrace();
            fail();
        }

        //we should have the same transaction back again
        assertEquals(tx, mgr.getTransaction());

        // there should be a registration for the remote gtx
        assertNotNull(table.get(remoteGtx));
        assertNotNull(table.getLocalTransaction(remoteGtx));


        List calls = dummy.getAllCalled();
        assertEquals(MethodDeclarations.optimisticPrepareMethod, calls.get(0));

        //assert we have two current transactions
        assertEquals(2, cache.getTransactionTable().getNumGlobalTransactions());
        assertEquals(2, cache.getTransactionTable().getNumLocalTransactions());

        //commit the local tx
        mgr.commit();

        //check local calls
        calls = dummy.getAllCalled();
        assertEquals(MethodDeclarations.optimisticPrepareMethod, calls.get(1));
        assertEquals(MethodDeclarations.commitMethod, calls.get(2));

        //assert we have only 1 transaction left

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

        assertNotNull(table.get(remoteGtx));
        assertNotNull(table.getLocalTransaction(remoteGtx));

        assertNull(table.get(gtx));
        assertNull(table.getLocalTransaction(gtx));
        //assert we are no longer associated
        assertEquals(null, mgr.getTransaction());
        cache.stopService();
    }

    public void testRemoteCommitSuspendTransaction() throws Exception
    {

        TreeCache cache = createCache();
        Interceptor txInterceptor = new TxInterceptor();
        txInterceptor.setCache(cache);
        Interceptor replicationInterceptor = new OptimisticReplicationInterceptor();
        replicationInterceptor.setCache(cache);
        Interceptor createInterceptor = new OptimisticCreateIfNotExistsInterceptor();
        createInterceptor.setCache(cache);
        Interceptor nodeInterceptor = new OptimisticNodeInterceptor();
        nodeInterceptor.setCache(cache);
        MockInterceptor dummy = new MockInterceptor();
        dummy.setCache(cache);

        txInterceptor.setNext(replicationInterceptor);
        replicationInterceptor.setNext(createInterceptor);
        createInterceptor.setNext(nodeInterceptor);
        nodeInterceptor.setNext(dummy);

        cache.setInterceptorChain(txInterceptor);

        DummyTransactionManager mgr = DummyTransactionManager.getInstance();

        //start local transaction
        mgr.begin();
        Transaction tx = mgr.getTransaction();

        //this sets
        cache.getCurrentTransaction(tx);

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

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

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

        //now send the remote prepare

        GlobalTransaction remoteGtx = new GlobalTransaction();

        remoteGtx.setAddress(new Address()
        {
            public boolean isMulticastAddress()
            {

                return false;
            }

            public void readExternal(ObjectInput arg0)
            {
            }

            public void writeExternal(ObjectOutput arg0)
            {
            }

            public int compareTo(Object arg0)
            {

                return 0;
            }

            public int size()
            {

                return 0;
            }

            public void writeTo(DataOutputStream arg0)
            {
            }

            public void readFrom(DataInputStream arg0)
            {
            }
        });
        //hack the method call to make it have the remote gtx
        MethodCall meth = (MethodCall) entry.getModifications().get(0);

        meth.getArgs()[0] = remoteGtx;
        //call our remote method
        MethodCall prepareMethod = MethodCallFactory.create(MethodDeclarations.optimisticPrepareMethod, new Object[]{remoteGtx, entry.getModifications(), null, remoteGtx.getAddress(), Boolean.FALSE});
        try
        {
            cache._replicate(prepareMethod);
        }
        catch (Throwable t)
        {
            fail();
        }

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

//		    call our remote method
        MethodCall commitMethod = MethodCallFactory.create(MethodDeclarations.commitMethod, new Object[]{remoteGtx});
        try
        {
            cache._replicate(commitMethod);
        }
        catch (Throwable t)
        {
            t.printStackTrace();
            fail();
        }

        //we should have the same transaction back again
        assertEquals(tx, mgr.getTransaction());

        //	 there should be a registration for the remote gtx
        assertNull(table.get(remoteGtx));
        assertNull(table.getLocalTransaction(remoteGtx));

        List calls = dummy.getAllCalled();
        assertEquals(MethodDeclarations.optimisticPrepareMethod, calls.get(0));
        assertEquals(MethodDeclarations.commitMethod, calls.get(1));

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

        //commit the local tx
        mgr.commit();

        calls = dummy.getAllCalled();
        assertEquals(MethodDeclarations.optimisticPrepareMethod, calls.get(2));
        assertEquals(MethodDeclarations.commitMethod, calls.get(3));


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

        assertEquals(null, mgr.getTransaction());
        cache.stopService();
    }

    public void testRemoteRollbackSuspendTransaction() throws Exception
    {

        TreeCache cache = createCache();
        Interceptor txInterceptor = new TxInterceptor();
        txInterceptor.setCache(cache);
        Interceptor replicationInterceptor = new OptimisticReplicationInterceptor();
        replicationInterceptor.setCache(cache);
        Interceptor createInterceptor = new OptimisticCreateIfNotExistsInterceptor();
        createInterceptor.setCache(cache);
        Interceptor nodeInterceptor = new OptimisticNodeInterceptor();
        nodeInterceptor.setCache(cache);
        MockInterceptor dummy = new MockInterceptor();
        dummy.setCache(cache);

        txInterceptor.setNext(replicationInterceptor);
        replicationInterceptor.setNext(createInterceptor);
        createInterceptor.setNext(nodeInterceptor);
        nodeInterceptor.setNext(dummy);

        cache.setInterceptorChain(txInterceptor);

        DummyTransactionManager mgr = DummyTransactionManager.getInstance();

        //start local transaction
        mgr.begin();
        Transaction tx = mgr.getTransaction();

        //this sets
        cache.getCurrentTransaction(tx);

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

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

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

        //now send the remote prepare

        GlobalTransaction remoteGtx = new GlobalTransaction();

        remoteGtx.setAddress(new Address()
        {
            public boolean isMulticastAddress()
            {

                return false;
            }

            public void readExternal(ObjectInput arg0)
            {
            }

            public void writeExternal(ObjectOutput arg0)
            {
            }

            public int compareTo(Object arg0)
            {

                return 0;
            }

            public int size()
            {

                return 0;
            }

            public void writeTo(DataOutputStream arg0)
            {
            }

            public void readFrom(DataInputStream arg0)
            {
            }
        });
        //hack the method call to make it have the remote gtx
        MethodCall meth = (MethodCall) entry.getModifications().get(0);

        meth.getArgs()[0] = remoteGtx;
        //call our remote method
        MethodCall prepareMethod = MethodCallFactory.create(MethodDeclarations.optimisticPrepareMethod, new Object[]{remoteGtx, entry.getModifications(), null, remoteGtx.getAddress(), Boolean.FALSE});
        try
        {
            cache._replicate(prepareMethod);
        }
        catch (Throwable t)
        {
            fail();
        }

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

//		    call our remote method
        MethodCall rollbackMethod = MethodCallFactory.create(MethodDeclarations.rollbackMethod, new Object[]{remoteGtx});
        try
        {
            cache._replicate(rollbackMethod);
        }
        catch (Throwable t)
        {
            fail();
        }

        //we should have the same transaction back again
        assertEquals(tx, mgr.getTransaction());

        //	 there should be a registration for the remote gtx
        assertNull(table.get(remoteGtx));
        assertNull(table.getLocalTransaction(remoteGtx));

        List calls = dummy.getAllCalled();
        assertEquals(MethodDeclarations.optimisticPrepareMethod, calls.get(0));
        assertEquals(MethodDeclarations.rollbackMethod, calls.get(1));

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

        //commit the local tx
        mgr.commit();

        calls = dummy.getAllCalled();
        assertEquals(MethodDeclarations.optimisticPrepareMethod, calls.get(2));
        assertEquals(MethodDeclarations.commitMethod, calls.get(3));


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

        assertEquals(null, mgr.getTransaction());
        cache.stopService();
    }

    public void testRemoteCommitTransaction() throws Exception
    {

        TreeCache cache = createCache();
        Interceptor txInterceptor = new TxInterceptor();
        txInterceptor.setCache(cache);
        Interceptor replicationInterceptor = new OptimisticReplicationInterceptor();
        replicationInterceptor.setCache(cache);
        Interceptor createInterceptor = new OptimisticCreateIfNotExistsInterceptor();
        createInterceptor.setCache(cache);
        Interceptor nodeInterceptor = new OptimisticNodeInterceptor();
        nodeInterceptor.setCache(cache);
        MockInterceptor dummy = new MockInterceptor();
        dummy.setCache(cache);

        txInterceptor.setNext(replicationInterceptor);
        replicationInterceptor.setNext(createInterceptor);
        createInterceptor.setNext(nodeInterceptor);
        nodeInterceptor.setNext(dummy);

        cache.setInterceptorChain(txInterceptor);

        DummyTransactionManager mgr = DummyTransactionManager.getInstance();

        //start local transaction
        mgr.begin();
        Transaction tx = mgr.getTransaction();

        //this sets
        cache.getCurrentTransaction(tx);

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

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

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

        //test local calls
        List calls = dummy.getAllCalled();
        assertEquals(MethodDeclarations.optimisticPrepareMethod, calls.get(0));
        assertEquals(MethodDeclarations.commitMethod, calls.get(1));

        assertNull(mgr.getTransaction());

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


        GlobalTransaction remoteGtx = new GlobalTransaction();

        remoteGtx.setAddress(new Address()
        {
            public boolean isMulticastAddress()
            {

                return false;
            }

            public void readExternal(ObjectInput arg0)
            {
            }

            public void writeExternal(ObjectOutput arg0)
            {
            }

            public int compareTo(Object arg0)
            {

                return 0;
            }

            public int size()
            {

                return 0;
            }

            public void writeTo(DataOutputStream arg0)
            {
            }

            public void readFrom(DataInputStream arg0)
            {
            }
        });

//	    hack the method call to make it have the remote gtx
        MethodCall meth = (MethodCall) entry.getModifications().get(0);

        meth.getArgs()[0] = remoteGtx;
        //call our remote method
        MethodCall prepareMethod = MethodCallFactory.create(MethodDeclarations.optimisticPrepareMethod, new Object[]{remoteGtx, entry.getModifications(), null, remoteGtx.getAddress(), Boolean.FALSE});
        try
        {
            cache._replicate(prepareMethod);
        }
        catch (Throwable t)
        {
            fail();
        }

        //our thread should be null
        assertNull(mgr.getTransaction());
        assertEquals(1, cache.getTransactionTable().getNumGlobalTransactions());
        assertEquals(1, cache.getTransactionTable().getNumLocalTransactions());

        //	 there should be a registration for the remote gtx
        assertNotNull(table.get(remoteGtx));
        assertNotNull(table.getLocalTransaction(remoteGtx));
        //this is not populated until replication interceptor is used
        assertEquals(1, table.get(remoteGtx).getModifications().size());

        calls = dummy.getAllCalled();
        assertEquals(MethodDeclarations.optimisticPrepareMethod, calls.get(2));

        assertNull(mgr.getTransaction());
//	    call our remote method
        MethodCall commitMethod = MethodCallFactory.create(MethodDeclarations.commitMethod, new Object[]{remoteGtx, Boolean.TRUE});
        try
        {
            cache._replicate(commitMethod);
        }
        catch (Throwable t)
        {
            fail();
        }

        assertNull(table.get(remoteGtx));
        assertNull(table.getLocalTransaction(remoteGtx));

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

    }

    public void testRemoteRollbackTransaction() throws Exception
    {

        TreeCache cache = createCache();
        Interceptor txInterceptor = new TxInterceptor();
        txInterceptor.setCache(cache);
        Interceptor replicationInterceptor = new OptimisticReplicationInterceptor();
        replicationInterceptor.setCache(cache);
        Interceptor createInterceptor = new OptimisticCreateIfNotExistsInterceptor();
        createInterceptor.setCache(cache);
        Interceptor nodeInterceptor = new OptimisticNodeInterceptor();
        nodeInterceptor.setCache(cache);
        MockInterceptor dummy = new MockInterceptor();
        dummy.setCache(cache);

        txInterceptor.setNext(replicationInterceptor);
        replicationInterceptor.setNext(createInterceptor);
        createInterceptor.setNext(nodeInterceptor);
        nodeInterceptor.setNext(dummy);

        cache.setInterceptorChain(txInterceptor);

        DummyTransactionManager mgr = DummyTransactionManager.getInstance();

        //start local transaction
        mgr.begin();
        Transaction tx = mgr.getTransaction();

        //this sets
        cache.getCurrentTransaction(tx);

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

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

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

        //test local calls
        List calls = dummy.getAllCalled();
        assertEquals(MethodDeclarations.optimisticPrepareMethod, calls.get(0));
        assertEquals(MethodDeclarations.commitMethod, calls.get(1));

        GlobalTransaction remoteGtx = new GlobalTransaction();

        remoteGtx.setAddress(new Address()
        {
            public boolean isMulticastAddress()
            {

                return false;
            }

            public void readExternal(ObjectInput arg0)
            {
            }

            public void writeExternal(ObjectOutput arg0)
            {
            }

            public int compareTo(Object arg0)
            {

                return 0;
            }

            public int size()
            {

                return 0;
            }

            public void writeTo(DataOutputStream arg0)
            {
            }

            public void readFrom(DataInputStream arg0)
            {
            }
        });

//	    hack the method call to make it have the remote gtx
        MethodCall meth = (MethodCall) entry.getModifications().get(0);

        meth.getArgs()[0] = remoteGtx;
        //call our remote method
        MethodCall prepareMethod = MethodCallFactory.create(MethodDeclarations.optimisticPrepareMethod, new Object[]{remoteGtx, entry.getModifications(), null, remoteGtx.getAddress(), Boolean.FALSE});
        try
        {
            cache._replicate(prepareMethod);
        }
        catch (Throwable t)
        {
            fail();
        }

        //our thread should be null
        assertNull(mgr.getTransaction());

        //	 there should be a registration for the remote gtx
        assertNotNull(table.get(remoteGtx));
        assertNotNull(table.getLocalTransaction(remoteGtx));
        //this is not populated until replication interceptor is used
        assertEquals(1, table.get(remoteGtx).getModifications().size());

        calls = dummy.getAllCalled();
        assertEquals(MethodDeclarations.optimisticPrepareMethod, calls.get(2));

//	    call our remote method
        MethodCall rollbackMethod = MethodCallFactory.create(MethodDeclarations.rollbackMethod, new Object[]{remoteGtx});
        try
        {
            cache._replicate(rollbackMethod);
        }
        catch (Throwable t)
        {
            fail();
        }

        calls = dummy.getAllCalled();
        assertEquals(MethodDeclarations.optimisticPrepareMethod, calls.get(2));
        assertEquals(MethodDeclarations.rollbackMethod, calls.get(3));

        assertNull(table.get(remoteGtx));
        assertNull(table.getLocalTransaction(remoteGtx));

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

    }


    public void testSequentialTransactionExists() throws Exception
    {

        TreeCache cache = createCache();
        Interceptor txInterceptor = new TxInterceptor();
        txInterceptor.setCache(cache);
        Interceptor createInterceptor = new OptimisticCreateIfNotExistsInterceptor();
        createInterceptor.setCache(cache);
        Interceptor nodeInterceptor = new OptimisticNodeInterceptor();
        nodeInterceptor.setCache(cache);
        MockInterceptor dummy = new MockInterceptor();
        dummy.setCache(cache);

        txInterceptor.setNext(createInterceptor);
        createInterceptor.setNext(nodeInterceptor);
        nodeInterceptor.setNext(dummy);

        cache.setInterceptorChain(txInterceptor);

        DummyTransactionManager mgr = DummyTransactionManager.getInstance();

        mgr.begin();
        Transaction tx = mgr.getTransaction();
        SamplePojo pojo = new SamplePojo(21, "test");

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

        assertNotNull(mgr.getTransaction());
        mgr.commit();


        mgr.begin();
        Transaction tx1 = mgr.getTransaction();
        assertNotNull(tx1);
        assertNotSame(tx, tx1);
        cache.put("/one/two", "key1", pojo);
        mgr.commit();


        List calls = dummy.getAllCalled();
        assertEquals(MethodDeclarations.optimisticPrepareMethod, calls.get(0));
        assertEquals(MethodDeclarations.commitMethod, calls.get(1));

        assertEquals(MethodDeclarations.optimisticPrepareMethod, calls.get(2));
        assertEquals(MethodDeclarations.commitMethod, calls.get(3));
        cache.stopService();
    }

}
