/*
 * JBoss, Home of Professional Open Source
 *
 * Distributable under LGPL license.
 * See terms of license at gnu.org.
 */
package org.jboss.cache.loader;

import junit.framework.Assert;
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.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import javax.transaction.TransactionManager;
import java.io.File;

/**
 * Tests using cache loaders with replicating data
 *
 * @author <a href="mailto:manik@jboss.org">Manik Surtani (manik@jboss.org)</a>
 */
public class CacheLoaderWithReplicationTest extends AbstractCacheLoaderTestBase
{
    private TreeCache cache1, cache2;
    private String tmpLocation1 = System.getProperty("java.io.tmpdir", ".") + File.separator + "JBossCache-CacheLoaderWithReplicationTest1";
    private String tmpLocation2 = System.getProperty("java.io.tmpdir", ".") + File.separator + "JBossCache-CacheLoaderWithReplicationTest2";
    private File dir1 = new File(tmpLocation1);
    private File dir2 = new File(tmpLocation2);
    private Fqn fqn = Fqn.fromString("/a");
    private String key = "key";

    private static final Log log = LogFactory.getLog(CacheLoaderWithReplicationTest.class);


    public CacheLoaderWithReplicationTest()
    {
        if (!dir1.exists()) dir1.mkdirs();
        if (!dir2.exists()) dir2.mkdirs();

        log.debug("   System props dump: " + System.getProperties());
        log.debug("Using location for CL 1 : " + tmpLocation1);
        log.debug("Using location for CL 2 : " + tmpLocation2);
    }

    public void setUp() throws Exception
    {
        cache1 = new TreeCache();
        cache1.setCacheLoaderConfiguration(getSingleCacheLoaderConfig("", "org.jboss.cache.loader.bdbje.BdbjeCacheLoader", "location=" + tmpLocation1, false, true, false));
        cache1.setTransactionManagerLookupClass("org.jboss.cache.DummyTransactionManagerLookup");

        cache2 = new TreeCache();
        cache2.setCacheLoaderConfiguration(getSingleCacheLoaderConfig("", "org.jboss.cache.loader.bdbje.BdbjeCacheLoader", "location=" + tmpLocation2, false, true, false));
        cache2.setTransactionManagerLookupClass("org.jboss.cache.DummyTransactionManagerLookup");
    }

    public void tearDown() throws Exception
    {
        if (cache1 != null)
        {
            try
            {
                cache1.remove(fqn);
                cache1.getCacheLoader().remove(fqn);
                cache1.stopService();
            }
            finally
            {
                cache1 = null;
            }
        }

        if (cache2 != null)
        {
            try
            {
                cache2.remove(fqn);
                cache2.getCacheLoader().remove(fqn);
                cache2.stopService();
            }
            finally
            {
                cache2 = null;
            }

        }
    }

    public void testPessSyncRepl() throws Exception
    {
        cache1.setCacheMode(TreeCache.REPL_SYNC);
        cache2.setCacheMode(TreeCache.REPL_SYNC);

        cache1.startService();
        cache2.startService();

        Assert.assertNull(cache1.get(fqn, key));
        Assert.assertNull(cache2.get(fqn, key));


        CacheLoader loader1 = cache1.getCacheLoader();
        CacheLoader loader2 = cache2.getCacheLoader();

        TransactionManager mgr = cache1.getTransactionManager();
        mgr.begin();
        cache1.put(fqn, key, "value");

        Assert.assertEquals("value", cache1.get(fqn, key));
        Assert.assertNull(cache2.get(fqn, key));
        Assert.assertNull(loader1.get(fqn));
        Assert.assertNull(loader2.get(fqn));
        mgr.commit();

        Assert.assertEquals("value", cache1.get(fqn, key));
        Assert.assertEquals("value", cache2.get(fqn, key));
        Assert.assertEquals("value", loader1.get(fqn).get(key));
        Assert.assertEquals("value", loader2.get(fqn).get(key));

        mgr.begin();
        cache1.put(fqn, key, "value2");

        Assert.assertEquals("value2", cache1.get(fqn, key));
        Assert.assertEquals("value", cache2.get(fqn, key));
        Assert.assertEquals("value", loader1.get(fqn).get(key));
        Assert.assertEquals("value", loader2.get(fqn).get(key));

        mgr.rollback();

        Assert.assertEquals("value", cache1.get(fqn, key));
        Assert.assertEquals("value", cache2.get(fqn, key));
        Assert.assertEquals("value", loader1.get(fqn).get(key));
        Assert.assertEquals("value", loader2.get(fqn).get(key));

        // force clean up
        tearDown();
    }

    public void testPessAsyncRepl() throws Exception
    {
        cache1.setCacheMode(TreeCache.REPL_ASYNC);
        cache2.setCacheMode(TreeCache.REPL_ASYNC);

        cache1.startService();
        cache2.startService();

        CacheLoader loader1 = cache1.getCacheLoader();
        CacheLoader loader2 = cache2.getCacheLoader();

        // make sure everything is empty...
        Assert.assertNull(loader1.get(fqn));
        Assert.assertNull(loader2.get(fqn));
        Assert.assertNull(cache1.get(fqn));
        Assert.assertNull(cache2.get(fqn));

        TransactionManager mgr = cache1.getTransactionManager();
        mgr.begin();
        cache1.put(fqn, key, "value");

        Assert.assertEquals("value", cache1.get(fqn, key));
        Assert.assertNull(cache2.get(fqn, key));
        Assert.assertNull(loader1.get(fqn));
        Assert.assertNull(loader2.get(fqn));
        mgr.commit();

        TestingUtil.sleepThread(500);

        Assert.assertEquals("value", cache1.get(fqn, key));
        Assert.assertEquals("value", cache2.get(fqn, key));
        Assert.assertEquals("value", loader1.get(fqn).get(key));
        Assert.assertEquals("value", loader2.get(fqn).get(key));

        mgr.begin();
        cache1.put(fqn, key, "value2");

        Assert.assertEquals("value2", cache1.get(fqn, key));
        Assert.assertEquals("value", cache2.get(fqn, key));
        Assert.assertEquals("value", loader1.get(fqn).get(key));
        Assert.assertEquals("value", loader2.get(fqn).get(key));

        mgr.rollback();

        TestingUtil.sleepThread(500);

        Assert.assertEquals("value", cache1.get(fqn, key));
        Assert.assertEquals("value", cache2.get(fqn, key));
        Assert.assertEquals("value", loader1.get(fqn).get(key));
        Assert.assertEquals("value", loader2.get(fqn).get(key));

        // force clean up
        tearDown();
    }

    public void testOptSyncRepl() throws Exception
    {
        try
        {
        cache1.setCacheMode(TreeCache.REPL_SYNC);
        cache2.setCacheMode(TreeCache.REPL_SYNC);

        cache1.setNodeLockingScheme("OPTIMISTIC");
        cache2.setNodeLockingScheme("OPTIMISTIC");

        cache1.startService();
        cache2.startService();

        CacheLoader loader1 = cache1.getCacheLoader();
        CacheLoader loader2 = cache2.getCacheLoader();

        TransactionManager mgr = cache1.getTransactionManager();
        mgr.begin();
        cache1.put(fqn, key, "value");

        Assert.assertEquals("value", cache1.get(fqn, key));
        Assert.assertNull(cache2.get(fqn, key));
        Assert.assertNull(loader1.get(fqn));
        Assert.assertNull(loader2.get(fqn));
        mgr.commit();

        Assert.assertEquals("value", cache1.get(fqn, key));
        Assert.assertEquals("value", cache2.get(fqn, key));
        Assert.assertEquals("value", loader1.get(fqn).get(key));
        Assert.assertEquals("value", loader2.get(fqn).get(key));

        mgr.begin();
        cache1.put(fqn, key, "value2");

        Assert.assertEquals("value2", cache1.get(fqn, key));
        Assert.assertEquals("value", cache2.get(fqn, key));
        Assert.assertEquals("value", loader1.get(fqn).get(key));
        Assert.assertEquals("value", loader2.get(fqn).get(key));

        mgr.rollback();

        Assert.assertEquals("value", cache1.get(fqn, key));
        Assert.assertEquals("value", cache2.get(fqn, key));
        Assert.assertEquals("value", loader1.get(fqn).get(key));
        Assert.assertEquals("value", loader2.get(fqn).get(key));
        }
        catch (Exception e)
        {
            Assert.assertTrue("Caught exception " + e.getMessage(), false);
            e.printStackTrace();
        }

        // force clean up
        tearDown();
    }

    public void testOptAsyncRepl() throws Exception
    {
        cache1.setCacheMode(TreeCache.REPL_ASYNC);
        cache2.setCacheMode(TreeCache.REPL_ASYNC);

        cache1.setNodeLockingScheme("OPTIMISTIC");
        cache2.setNodeLockingScheme("OPTIMISTIC");

        cache1.startService();
        cache2.startService();

        CacheLoader loader1 = cache1.getCacheLoader();
        CacheLoader loader2 = cache2.getCacheLoader();

        TransactionManager mgr = cache1.getTransactionManager();
        mgr.begin();
        cache1.put(fqn, key, "value");

        Assert.assertEquals("value", cache1.get(fqn, key));
        Assert.assertNull(cache2.get(fqn, key));
        Assert.assertNull(loader1.get(fqn));
        Assert.assertNull(loader2.get(fqn));
        mgr.commit();

        TestingUtil.sleepThread(500);

        Assert.assertEquals("value", cache1.get(fqn, key));
        Assert.assertEquals("value", cache2.get(fqn, key));
        Assert.assertEquals("value", loader1.get(fqn).get(key));
        Assert.assertEquals("value", loader2.get(fqn).get(key));

        mgr.begin();
        cache1.put(fqn, key, "value2");

        Assert.assertEquals("value2", cache1.get(fqn, key));
        Assert.assertEquals("value", cache2.get(fqn, key));
        Assert.assertEquals("value", loader1.get(fqn).get(key));
        Assert.assertEquals("value", loader2.get(fqn).get(key));

        mgr.rollback();
        TestingUtil.sleepThread(500);

        Assert.assertEquals("value", cache1.get(fqn, key));
        Assert.assertEquals("value", cache2.get(fqn, key));
        Assert.assertEquals("value", loader1.get(fqn).get(key));
        Assert.assertEquals("value", loader2.get(fqn).get(key));

        // force clean up
        tearDown();
    }

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

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

}


