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

import org.jboss.cache.TreeCache;
import org.jboss.cache.Fqn;
import org.jboss.cache.config.Option;
import org.jboss.cache.misc.TestingUtil;

/**
 * Tests behaviour when data owners fail - essentially this tests data gravitation
 *
 * @author <a href="mailto:manik@jboss.org">Manik Surtani (manik@jboss.org)</a>
 */
public class BuddyReplicationFailoverTest extends BuddyReplicationTestsBase
{
    protected boolean optimisticLocks = false;
    private String key = "key";
    private String value = "value";
    private TreeCache[] caches;

    protected void tearDown()
    {
        if (caches != null)
        {
            cleanup(caches);
            caches = null;
        }
    }


    public void testDataGravitationKillOwner() throws Exception
    {
        testDataGravitation(true);
    }

    public void testDataGravitationDontKillOwner() throws Exception
    {
        testDataGravitation(false);
    }

    private void testDataGravitation(boolean killOwner) throws Exception
    {
        caches = createCaches(3, false, true, optimisticLocks);

        String fqn = "/test";
        String backupFqn = "/" + BuddyManager.BUDDY_BACKUP_SUBTREE + "/" + BuddyManager.getGroupNameFromAddress(caches[0].getLocalAddress()) + fqn;

        dumpCacheContents(caches);

        caches[0].put(fqn, key, value);

        dumpCacheContents(caches);

        assertEquals("Value should exist", value, caches[0].get(fqn, key));

        dumpCacheContents(caches);

        // use exists instead of get() to prevent going up the interceptor stack
        assertTrue("Should be false", !caches[1].exists(fqn));
        assertTrue("Should be false", !caches[2].exists(fqn));

        assertNull("Should be null", caches[0].get(backupFqn, key));
        assertEquals("Value should exist", value, caches[1].get(backupFqn, key));
        assertNull("Should be null", caches[2].get(backupFqn, key));

        if (killOwner)
        {
            caches[0].stopService();
            caches[0] = null;
            TestingUtil.sleepThread(500);
        }

        System.out.println("***** Killed original data owner, about to call a get on a different cache instance.  *****");

        // according to data gravitation, a call to *any* cache should retrieve the data, and move the data to the new cache.
        assertEquals("Value should have gravitated", value, caches[2].get(fqn, key));

        TestingUtil.sleepThread(500);

        dumpCacheContents(caches);
        
        // now lets test the eviction part of gravitation
        String newBackupFqn = "/" + BuddyManager.BUDDY_BACKUP_SUBTREE + "/" + BuddyManager.getGroupNameFromAddress(caches[2].getLocalAddress()) + fqn;

        // use exists instead of get() to prevent going up the interceptor stack
        if (!killOwner) assertTrue("Should be false", !caches[0].exists(fqn));
        assertTrue("Should be false", !caches[1].exists(fqn));

        // the old backup should no longer exist
        if (!killOwner) assertNull("Should be null", caches[0].get(backupFqn, key));
        assertNull("Should be null", caches[1].get(backupFqn, key));
        assertNull("Should be null", caches[2].get(backupFqn, key));

        // and the backup should now exist in caches[2]'s buddy which is caches[0]
        if (killOwner)
        {
            assertEquals("Value should exist", value, caches[1].get(newBackupFqn, key));
        }
        else
        {
            assertEquals("Value should exist", value, caches[0].get(newBackupFqn, key));
            assertNull("Should be null", caches[1].get(newBackupFqn, key));
        }
        assertNull("Should be null", caches[2].get(newBackupFqn, key));
    }

    public void testDataReplicationSuppression() throws Exception
    {
        caches = createCaches(3, false, false, optimisticLocks);

        Fqn fqn = Fqn.fromString("/test");
        Fqn backupFqn = Fqn.fromString("/" + BuddyManager.BUDDY_BACKUP_SUBTREE + "/" + BuddyManager.getGroupNameFromAddress(caches[0].getLocalAddress())
                + "/test");
        caches[0].put(fqn, key, value);

        dumpCacheContents(caches);

        assertEquals("value", caches[0].get(fqn, key));
        assertNull(caches[0].get(backupFqn, key));
        assertEquals("value", caches[1].get(backupFqn, key));
        assertNull(caches[1].get(fqn, key));
        assertNull(caches[2].get(fqn, key));
        assertNull(caches[2].get(backupFqn, key));
        
        assertNoLocks(caches);
        
        // Now force a gravitation
        Option force = new Option();
        force.setForceDataGravitation(true);
        backupFqn = Fqn.fromString("/" + BuddyManager.BUDDY_BACKUP_SUBTREE + "/" + BuddyManager.getGroupNameFromAddress(caches[1].getLocalAddress())
              + "/test");

        System.out.println("*** Calling get() on cache[1] with force option");

        assertEquals("value", caches[1].get(fqn, key, force));

        dumpCacheContents(caches);

        assertNull(caches[1].get(backupFqn, key));
        assertEquals("value", caches[2].get(backupFqn, key));
        assertNull(caches[2].get(fqn, key));
        assertNull(caches[0].get(fqn, key));
        assertNull(caches[0].get(backupFqn, key));
    }

    public void testSubtreeRetrieval() throws Exception
    {
        caches = createCaches(3, false, true, optimisticLocks);

        Fqn fqn = Fqn.fromString("/test");
        Fqn fqn2 = Fqn.fromString("/test/subtree");

        Fqn backupFqn = Fqn.fromString("/" + BuddyManager.BUDDY_BACKUP_SUBTREE + "/" + BuddyManager.getGroupNameFromAddress(caches[0].getLocalAddress()) + "/test");
        Fqn backupFqn2 = Fqn.fromString("/" + BuddyManager.BUDDY_BACKUP_SUBTREE + "/" + BuddyManager.getGroupNameFromAddress(caches[0].getLocalAddress()) + "/test/subtree");



        caches[0].put(fqn, key, value);
        caches[0].put(fqn2, key, value);

        // test backup replication to buddy
        assertEquals(value, caches[0].get(fqn, key));
        assertEquals(value, caches[0].get(fqn2, key));
        assertEquals(value, caches[1].get(backupFqn, key));
        assertEquals(value, caches[1].get(backupFqn2, key));

        assertTrue(!caches[0].exists(backupFqn));
        assertTrue(!caches[0].exists(backupFqn2));
        assertTrue(!caches[1].exists(fqn));
        assertTrue(!caches[1].exists(fqn2));
        assertTrue(!caches[2].exists(fqn));
        assertTrue(!caches[2].exists(fqn2));
        assertTrue(!caches[2].exists(backupFqn));
        assertTrue(!caches[2].exists(backupFqn2));

        assertNoLocks(caches);

        // gravitate to 2:
        caches[2].get(fqn);  // expect entire subtree to gravitate.

        Fqn newBackupFqn = Fqn.fromString("/" + BuddyManager.BUDDY_BACKUP_SUBTREE + "/" + BuddyManager.getGroupNameFromAddress(caches[2].getLocalAddress()) + "/test");
        Fqn newBackupFqn2 = Fqn.fromString("/" + BuddyManager.BUDDY_BACKUP_SUBTREE + "/" + BuddyManager.getGroupNameFromAddress(caches[2].getLocalAddress()) + "/test/subtree");

        assertEquals(value, caches[2].get(fqn, key));
        assertTrue(caches[2].exists(fqn2));
        assertEquals(value, caches[0].get(newBackupFqn, key));
        assertTrue(caches[0].exists(newBackupFqn2));

        assertTrue(!caches[2].exists(newBackupFqn));
        assertTrue(!caches[2].exists(newBackupFqn2));
        assertTrue(!caches[0].exists(fqn));
        assertTrue(!caches[0].exists(fqn2));
        assertTrue(!caches[1].exists(fqn));
        assertTrue(!caches[1].exists(fqn2));
        assertTrue(!caches[1].exists(newBackupFqn));
        assertTrue(!caches[1].exists(newBackupFqn2));

        for (int i=0; i<caches.length; i++)
        {
            assertTrue(!caches[i].exists(backupFqn));
            assertTrue(!caches[i].exists(backupFqn2));
        }

        assertNoLocks(caches);
    }
}
