/*
 * JBoss, Home of Professional Open Source
 * Copyright 2012 Red Hat Inc. and/or its affiliates and other contributors
 * as indicated by the @author tags. All rights reserved.
 * See the copyright.txt in the distribution for a
 * full listing of individual contributors.
 *
 * This copyrighted material is made available to anyone wishing to use,
 * modify, copy, or redistribute it subject to the terms and conditions
 * of the GNU Lesser General Public License, v. 2.1.
 * This program is distributed in the hope that it will be useful, but WITHOUT A
 * WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A
 * PARTICULAR PURPOSE.  See the GNU Lesser General Public License for more details.
 * You should have received a copy of the GNU Lesser General Public License,
 * v.2.1 along with this distribution; if not, write to the Free Software
 * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston,
 * MA  02110-1301, USA.
 */

package org.infinispan.tx;

import org.infinispan.configuration.cache.CacheMode;
import org.infinispan.configuration.cache.ConfigurationBuilder;
import org.infinispan.test.MultipleCacheManagersTest;
import org.infinispan.util.concurrent.IsolationLevel;
import org.testng.annotations.Test;

import javax.transaction.Transaction;

import static org.testng.AssertJUnit.assertEquals;

/**
 * @author Mircea Markus
 * @since 5.2
 */
@Test (groups = "functional", testName = "tx.ReadCommittedDistributedTest")
public class ReadCommittedDistributedTest extends MultipleCacheManagersTest {

   @Override
   protected void createCacheManagers() throws Throwable {
      ConfigurationBuilder dccc = getDefaultClusteredCacheConfig(CacheMode.DIST_SYNC, true);
      dccc.clustering().hash().numOwners(1);
      dccc.transaction().locking().isolationLevel(IsolationLevel.READ_COMMITTED);
      createCluster(dccc, 2);
   }

   public void testValueChangedOnSameNode() throws Throwable {
      Object keyNode0 = getKeyForCache(0);
      String v0 = "v0";
      cache(0).put(keyNode0, v0);
      assertEquals(v0, cache(0).get(keyNode0));

      //check that if the modification is on the same node then the transaction sees it
      String v1 = "v1";
      tm(0).begin();
      cache(0).put("kkk", "vvv");
      Object initialValue = cache(0).get(keyNode0);
      assertEquals(v0, initialValue);
      Transaction suspend = tm(0).suspend();

      cache(0).put(keyNode0, v1);
      tm(0).resume(suspend);
      assertEquals(v1, cache(0).get(keyNode0));
      tm(0).commit();

      assertEquals(v1, cache(0).get(keyNode0));
      assertEquals("vvv", cache(0).get("kkk"));
   }
   public void testValueChangedOnDifferentNode() throws Throwable {
      Object keyNode1 = getKeyForCache(1);
      String v0 = "v0";
      cache(0).put(keyNode1, v0);
      assertEquals(v0, cache(0).get(keyNode1));

      //check that if the modification is on the same node then the transaction sees it
      tm(0).begin();
      cache(0).put("kkk", "vvv");
      assertEquals(v0, cache(0).get(keyNode1));
      Transaction suspend = tm(0).suspend();

      String v1 = "v1";
      cache(1).put(keyNode1, v1);

      tm(0).resume(suspend);
      assertEquals(v1, cache(0).get(keyNode1));
      tm(0).commit();

      assertEquals(v1, cache(0).get(keyNode1));
      assertEquals("vvv", cache(0).get("kkk"));
   }
}