Tests loading from cache store using functional API

AbstractFunctionalCachestoreTest.java

package org.infinispan.functional;

import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertTrue;

import java.io.Serializable;
import java.util.Set;
import java.util.function.Function;

import javax.transaction.TransactionManager;

import org.infinispan.Cache;
import org.infinispan.commons.api.functional.EntryView.ReadWriteEntryView;
import org.infinispan.commons.api.functional.FunctionalMap.ReadWriteMap;
import org.infinispan.commons.api.functional.Param;
import org.infinispan.configuration.cache.CacheMode;
import org.infinispan.configuration.cache.ConfigurationBuilder;
import org.infinispan.functional.impl.ReadWriteMapImpl;
import org.infinispan.persistence.dummy.DummyInMemoryStore;
import org.infinispan.persistence.dummy.DummyInMemoryStoreConfigurationBuilder;
import org.infinispan.persistence.manager.PersistenceManager;
import org.infinispan.transaction.lookup.DummyTransactionManagerLookup;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;

/**
 * @author Radim Vansa <rvansa@redhat.com>
 */
public abstract class AbstractFunctionalCachestoreTest extends AbstractFunctionalTest {
   private final boolean tx;
   private ReadWriteMap<Object, String> rw;

   protected AbstractFunctionalCachestoreTest(boolean tx) {
      this.tx = tx;
   }

   @Override
   protected void createCacheManagers() throws Throwable {
      // Create local caches as default in a cluster of 2
      createClusteredCaches(2, new ConfigurationBuilder());
      // Create distributed caches
      ConfigurationBuilder distBuilder = new ConfigurationBuilder();
      distBuilder.clustering().cacheMode(CacheMode.DIST_SYNC).hash().numOwners(1)
                 .persistence().addStore(DummyInMemoryStoreConfigurationBuilder.class);
      if (tx) {
         distBuilder.transaction().transactionManagerLookup(new DummyTransactionManagerLookup());
      }
      cacheManagers.stream().forEach(cm -> cm.defineConfiguration(DIST, distBuilder.build()));
      // Create replicated caches
      ConfigurationBuilder replBuilder = new ConfigurationBuilder();
      replBuilder.clustering().cacheMode(CacheMode.REPL_SYNC)
                 .persistence().addStore(DummyInMemoryStoreConfigurationBuilder.class);
      if (tx) {
         replBuilder.transaction().transactionManagerLookup(new DummyTransactionManagerLookup());
      }
      cacheManagers.stream().forEach(cm -> cm.defineConfiguration(REPL, replBuilder.build()));
      // Wait for cluster to form
      waitForClusterToForm(DIST, REPL);
   }

   @Override
   @BeforeClass
   public void createBeforeClass() throws Throwable {
      super.createBeforeClass();
      rw = ReadWriteMapImpl.create(fmapD1);
   }

   @Test
   public void testLoadSame() throws Exception {
      doTest(0);
   }

   @Test
   public void testLoadRemote() throws Exception {
      doTest(1);
   }

   private void doTest(int targetNode) throws Exception {
      Cache<Object, String> cache = cache(targetNode, DIST);
      TransactionManager tm = cache.getAdvancedCache().getTransactionManager();
      Object key = getKeyForCache(targetNode, DIST);

      if (tx) {
         tm.begin();
      }
      rw.withParams(Param.FutureMode.COMPLETED).eval(key, (Function<ReadWriteEntryView<Object, String>, Object> & Serializable) view -> {
         assertFalse(view.find().isPresent());
         view.set("value");
         return null;
      }).get();
      if (tx) {
         tm.commit();
      }

      assertEquals(cache.get(key), "value");
      cache.evict(key);
      assertFalse(cache.getAdvancedCache().getDataContainer().containsKey(key));
      Set<DummyInMemoryStore> stores = cache.getAdvancedCache().getComponentRegistry().getComponent(PersistenceManager.class).getStores(DummyInMemoryStore.class);
      DummyInMemoryStore store = stores.iterator().next();
      assertTrue(store.contains(key));

      if (tx) {
         tm.begin();
      }
      rw.withParams(Param.FutureMode.COMPLETED).eval(key, (Function<ReadWriteEntryView<Object, String>, Object> & Serializable) view -> {
         assertEquals(view.get(), "value");
         assertTrue(view.find().isPresent());
         return null;
      }).get();
      if (tx) {
         tm.commit();
      }
   }
}

FunctionalCachestoreTestNonTx.java

package org.infinispan.functional;

public class FunctionalCachestoreTestNonTx extends AbstractFunctionalCachestoreTest {
   public FunctionalCachestoreTestNonTx() {
      super(false);
   }
}

FunctionalCachestoreTestTx.java

package org.infinispan.functional;

public class FunctionalCachestoreTestTx extends AbstractFunctionalCachestoreTest {
   public FunctionalCachestoreTestTx() {
      super(true);
   }
}